diff --git a/app/config/settings.py b/app/config/settings.py
index 308b651..a30cd57 100644
--- a/app/config/settings.py
+++ b/app/config/settings.py
@@ -24,6 +24,10 @@ class Settings(BaseSettings):
     cors_allowed_origins: str = Field(
         default="", json_schema_extra={"env": "CORS_ALLOWED_ORIGINS"}
     )
+    dispatch_api_public_url: str = Field(
+        default="http://localhost:8000",
+        json_schema_extra={"env": "DISPATCH_API_PUBLIC_URL"},
+    )
 
     model_config = SettingsConfigDict(
         env_file=".env",
diff --git a/app/main.py b/app/main.py
index 5a596a7..fae83a3 100644
--- a/app/main.py
+++ b/app/main.py
@@ -4,6 +4,7 @@
 from app.middleware.correlation_id import add_correlation_id
 from app.middleware.error_handling import register_exception_handlers
 from app.platforms.dispatcher import load_processing_platforms
+from app.platforms.implementations.gep.router import router as gep_callbacks_router
 from app.services.tiles.base import load_grids
 from app.config.logger import setup_logging
 from app.config.settings import settings
@@ -59,3 +60,5 @@ async def root():
 app.include_router(health.router)
 app.include_router(parameters.router)
 app.include_router(statistics.router)
+
+app.include_router(gep_callbacks_router)
diff --git a/app/middleware/error_handling.py b/app/middleware/error_handling.py
index b63b4e8..dd116b3 100644
--- a/app/middleware/error_handling.py
+++ b/app/middleware/error_handling.py
@@ -1,6 +1,10 @@
 import traceback
 from typing import Any
 
+from http_apis_problems_detail.fastapi import (
+    ProblemRegistryException,
+    problem_registry_exception_handler,
+)
 from fastapi import Request, status
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
@@ -83,6 +87,7 @@ def register_exception_handlers(app):
     Call this in main.py after creating the FastAPI() instance.
     """
 
+    app.add_exception_handler(ProblemRegistryException, problem_registry_exception_handler)
     app.add_exception_handler(DispatcherException, dispatch_exception_handler)
     app.add_exception_handler(RequestValidationError, validation_exception_handler)
     app.add_exception_handler(Exception, generic_exception_handler)
diff --git a/app/platforms/implementations/gep/__init__.py b/app/platforms/implementations/gep/__init__.py
new file mode 100644
index 0000000..6e6aa00
--- /dev/null
+++ b/app/platforms/implementations/gep/__init__.py
@@ -0,0 +1 @@
+"""GEP stage-event callback integration for OGC API Processes."""
diff --git a/app/platforms/implementations/gep/auth.py b/app/platforms/implementations/gep/auth.py
new file mode 100644
index 0000000..7714187
--- /dev/null
+++ b/app/platforms/implementations/gep/auth.py
@@ -0,0 +1,92 @@
+"""Authenticate GEP callback service accounts independently of APEx user login."""
+
+from functools import lru_cache
+from typing import Annotated
+
+import jwt
+from fastapi import Depends, Request
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from http_apis_problems_detail import ErrorDetail
+from http_apis_problems_detail.fastapi import ForbiddenException, UnauthorizedException
+from jwt import PyJWKClient
+
+from app.platforms.implementations.gep.settings import get_callback_backends
+
+callback_bearer = HTTPBearer(auto_error=False, scheme_name="GepCallbackBearer")
+
+
+@lru_cache(maxsize=32)
+def _callback_jwks_client(issuer: str) -> PyJWKClient:
+    # Called only after matching a configured issuer, never an arbitrary token URL.
+    # Cache the JWKS rather than individual keys indefinitely, allowing key rotation.
+    return PyJWKClient(
+        f"{issuer.rstrip('/')}/protocol/openid-connect/certs",
+        cache_jwk_set=True,
+        lifespan=300,
+        timeout=10,
+    )
+
+
+def verify_callback_token(
+    request: Request,
+    creds: Annotated[HTTPAuthorizationCredentials | None, Depends(callback_bearer)],
+) -> set[str]:
+    """Return the backend endpoints this verified service account may update."""
+    unauthorized = UnauthorizedException(
+        request=request,
+        errors=[ErrorDetail(
+            header="Authorization", detail="Callback Bearer Authorization token not set or invalid."
+        )],
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    if creds is None or creds.scheme.lower() != "bearer":
+        raise unauthorized
+
+    # Configuration errors are server errors, not invalid client credentials.
+    configured_backends = get_callback_backends()
+    try:
+        issuer = jwt.decode(
+            creds.credentials, options={"verify_signature": False}
+        ).get("iss")
+        candidates = {
+            endpoint: cfg for endpoint, cfg in configured_backends.items()
+            if isinstance(issuer, str) and issuer
+            and cfg.callback_issuer == issuer
+            and cfg.callback_audience and cfg.callback_clients
+        }
+        if not candidates:
+            raise unauthorized
+
+        key = _callback_jwks_client(issuer).get_signing_key_from_jwt(creds.credentials).key
+        allowed: set[str] = set()
+        valid_audience = False
+        # Require an audience match; reject any other JWT validation failure.
+        for endpoint, cfg in candidates.items():
+            try:
+                claims = jwt.decode(
+                    creds.credentials,
+                    key,
+                    algorithms=["RS256"],
+                    issuer=issuer,
+                    audience=cfg.callback_audience,
+                    options={"require": ["iss", "aud", "exp"]},
+                )
+            except jwt.InvalidAudienceError:
+                continue
+            valid_audience = True
+            client = claims.get("azp")
+            if isinstance(client, str) and client in cfg.callback_clients:
+                allowed.add(endpoint.rstrip("/"))
+        if not valid_audience:
+            raise unauthorized
+    except (jwt.PyJWTError, ValueError, TypeError, OverflowError) as exc:
+        raise unauthorized from exc
+
+    if not allowed:
+        raise ForbiddenException(
+            request=request,
+            errors=[ErrorDetail(
+                header="Authorization", detail="Callback client is not allowed."
+            )],
+        )
+    return allowed
diff --git a/app/platforms/implementations/gep/callbacks.py b/app/platforms/implementations/gep/callbacks.py
new file mode 100644
index 0000000..86e5059
--- /dev/null
+++ b/app/platforms/implementations/gep/callbacks.py
@@ -0,0 +1,49 @@
+"""GEP event interpretation and exact platform-job correlation."""
+
+from sqlalchemy.orm import Session
+from stage_events_client.models import BaseCloudEvent
+
+from app.database.models.processing_job import ProcessingJobRecord
+from app.schemas.enum import ProcessTypeEnum, ProcessingStatusEnum
+
+STAGE_TO_APEX_STATUS: dict[str, ProcessingStatusEnum] = {
+    "submitted": ProcessingStatusEnum.QUEUED,
+    "prepared": ProcessingStatusEnum.RUNNING,
+    "staged": ProcessingStatusEnum.RUNNING,
+    "piped": ProcessingStatusEnum.FINISHED,
+    "completed": ProcessingStatusEnum.FINISHED,
+    "failed": ProcessingStatusEnum.FAILED,
+    "dismissed": ProcessingStatusEnum.CANCELED,
+}
+
+
+def extract_job_identifier(cloud_event: BaseCloudEvent) -> str | None:
+    """Extract job_identifier strictly from CloudEvent subject metadata."""
+    # CloudEvent subject pattern: "<namespace>:<workflow_uid>:<workflow_name>"
+    subject = getattr(cloud_event, "subject", None)
+    if not subject:
+        return None
+
+    parts = subject.split(":")
+    if len(parts) == 3:
+        return f"{parts[0]}:{parts[2]}"
+    return subject
+
+
+def get_gep_job(database: Session, job_identifier: str) -> ProcessingJobRecord | None:
+    """Never interpret a GEP identifier as an APEx ID or use suffix matching.
+
+    Subjects carry no trusted backend identity. If multiple OGC jobs have
+    overlapping identifiers, reject the event instead of picking the first job.
+    """
+    matches = (
+        database.query(ProcessingJobRecord)
+        .filter(
+            ProcessingJobRecord.label == ProcessTypeEnum.OGC_API_PROCESS,
+            ProcessingJobRecord.platform_job_id == job_identifier,
+        )
+        .all()
+    )
+    if len(matches) > 1:
+        raise ValueError("Ambiguous GEP job identifier")
+    return matches[0] if matches else None
diff --git a/app/platforms/implementations/gep/router.py b/app/platforms/implementations/gep/router.py
new file mode 100644
index 0000000..356dcc1
--- /dev/null
+++ b/app/platforms/implementations/gep/router.py
@@ -0,0 +1,209 @@
+"""Receive GEP stage CloudEvents, not arbitrary OGC callback payloads."""
+
+import json
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Request, status
+from loguru import logger
+from sqlalchemy.orm import Session
+from stage_events_client.models import BaseCloudEvent
+from http_apis_problems_detail import ErrorDetail
+from http_apis_problems_detail.fastapi import (
+    PROBLEM_RESPONSES, BusinessRuleViolationException, ServerErrorException,
+    NotFoundException, ProblemRegistryException, ForbiddenException,
+)
+
+from app.platforms.implementations.gep.auth import verify_callback_token
+from app.database.db import get_db
+from app.schemas.enum import ProcessingStatusEnum
+from app.services.job_status import apply_job_status, InvalidJobTransition
+from app.platforms.implementations.gep.callbacks import (
+    STAGE_TO_APEX_STATUS, extract_job_identifier, get_gep_job,
+)
+from app.platforms.implementations.gep.schemas import (
+    ProgressEvent, SuccessEvent, FailureEvent,
+)
+from app.platforms.implementations.gep.subscriber import CALLBACK_PREFIX
+
+router = APIRouter(prefix=CALLBACK_PREFIX, tags=["GEP callbacks"])
+
+
+def _handle_callback_status_update(
+    db: Session,
+    cloud_event: BaseCloudEvent,
+    target_status: ProcessingStatusEnum,
+    request: Request,
+    caller: set[str],
+) -> None:
+    """Internal helper to update job status from callback CloudEvents."""
+    job_identifier = extract_job_identifier(cloud_event) or ""
+    endpoint_name = request.url.path.split("/")[-1]
+
+    logger.info(f"Callback '{endpoint_name}' received for job '{job_identifier}'")
+
+    try:
+        job = get_gep_job(db, job_identifier)
+        if not job:
+            if endpoint_name == "inProgress":
+                # An early inProgress callback may arrive before initial job commit finishes;
+                # acknowledge gracefully without returning 404.
+                logger.debug(
+                    f"Job '{job_identifier}' is not registered in database yet; "
+                    "acknowledging in progress"
+                )
+                return
+
+            raise NotFoundException(
+                request=request,
+                errors=[
+                    ErrorDetail(
+                        pointer="subject",
+                        detail=f"Job {job_identifier} does not exist in the Job registry",
+                    )
+                ]
+            )
+
+        # Authorize against persisted dispatch metadata, not CloudEvent claims.
+        endpoint = json.loads(job.service)["endpoint"].rstrip("/")
+        if endpoint not in caller:
+            raise ForbiddenException(
+                request=request,
+                errors=[ErrorDetail(
+                    pointer="subject",
+                    detail="Callback caller is not authorized to update this job.",
+                )],
+            )
+
+        try:
+            apply_job_status(db, job, target_status)
+        except InvalidJobTransition as exc:
+            raise BusinessRuleViolationException(
+                request=request,
+                errors=[ErrorDetail(pointer="subject", detail=str(exc))],
+            ) from exc
+        logger.info(
+            f"Successfully updated status of job '{job_identifier}' (id={job.id}) "
+            f"to {target_status}"
+        )
+    except ProblemRegistryException:
+        raise
+    except Exception as e:
+        logger.error(f"Error handling {endpoint_name} callback for job {job_identifier}: {e}")
+        raise ServerErrorException(
+            request=request,
+            errors=[
+                ErrorDetail(
+                    parameter="job_id",
+                    detail=(
+                        f"An error occurred when updating status of Job {job_identifier}: {e}"
+                    ),
+                )
+            ]
+        )
+
+
+@router.post(
+    "/success",
+    response_model=None,
+    status_code=status.HTTP_200_OK,
+    summary="URIs for success callbacks for jobs.",
+    description=(
+        "Accepts GEP stage CloudEvents. The callback URL will be invoked "
+        "after a job succeeded its execution.\n"
+    ),
+    operation_id="onGepJobSuccess",
+    responses={
+        status.HTTP_200_OK: {"description": "Job status updated successfully."},
+        status.HTTP_401_UNAUTHORIZED: PROBLEM_RESPONSES[status.HTTP_401_UNAUTHORIZED],
+        status.HTTP_403_FORBIDDEN: PROBLEM_RESPONSES[status.HTTP_403_FORBIDDEN],
+        status.HTTP_400_BAD_REQUEST: PROBLEM_RESPONSES[status.HTTP_400_BAD_REQUEST],
+        status.HTTP_404_NOT_FOUND: PROBLEM_RESPONSES[status.HTTP_404_NOT_FOUND],
+        status.HTTP_422_UNPROCESSABLE_CONTENT: (
+            PROBLEM_RESPONSES[status.HTTP_422_UNPROCESSABLE_CONTENT]
+        ),
+        status.HTTP_500_INTERNAL_SERVER_ERROR: (
+            PROBLEM_RESPONSES[status.HTTP_500_INTERNAL_SERVER_ERROR]
+        ),
+    },
+)
+def on_gep_job_success(
+    request: Request,
+    body: SuccessEvent,
+    caller: Annotated[set[str], Depends(verify_callback_token)],
+    db: Session = Depends(get_db),
+) -> None:
+    """URIs for success callbacks for jobs."""
+    _handle_callback_status_update(
+        db, body, ProcessingStatusEnum.FINISHED, request, caller
+    )
+
+
+@router.post(
+    "/inProgress",
+    response_model=None,
+    status_code=status.HTTP_200_OK,
+    summary="URIs for progress callbacks for jobs.",
+    description=(
+        "Accepts GEP stage CloudEvents. The callback URL will be invoked while a job is changing "
+        "its status during its execution.\n"
+    ),
+    operation_id="onGepJobProgress",
+    responses={
+        status.HTTP_200_OK: {"description": "Job status updated successfully."},
+        status.HTTP_401_UNAUTHORIZED: PROBLEM_RESPONSES[status.HTTP_401_UNAUTHORIZED],
+        status.HTTP_403_FORBIDDEN: PROBLEM_RESPONSES[status.HTTP_403_FORBIDDEN],
+        status.HTTP_400_BAD_REQUEST: PROBLEM_RESPONSES[status.HTTP_400_BAD_REQUEST],
+        status.HTTP_404_NOT_FOUND: PROBLEM_RESPONSES[status.HTTP_404_NOT_FOUND],
+        status.HTTP_422_UNPROCESSABLE_CONTENT: (
+            PROBLEM_RESPONSES[status.HTTP_422_UNPROCESSABLE_CONTENT]
+        ),
+        status.HTTP_500_INTERNAL_SERVER_ERROR: (
+            PROBLEM_RESPONSES[status.HTTP_500_INTERNAL_SERVER_ERROR]
+        ),
+    },
+)
+def on_gep_job_progress(
+    request: Request,
+    body: ProgressEvent,
+    caller: Annotated[set[str], Depends(verify_callback_token)],
+    db: Session = Depends(get_db),
+) -> None:
+    """URIs for progress callbacks for jobs."""
+    target_status = STAGE_TO_APEX_STATUS[body.type]
+    _handle_callback_status_update(db, body, target_status, request, caller)
+
+
+@router.post(
+    "/failed",
+    response_model=None,
+    status_code=status.HTTP_200_OK,
+    summary="URIs for failed callbacks for jobs.",
+    description=(
+        "Accepts GEP stage CloudEvents. The callback URL will be invoked "
+        "after a job failed its execution.\n"
+    ),
+    operation_id="onGepJobFailed",
+    responses={
+        status.HTTP_200_OK: {"description": "Job status updated successfully."},
+        status.HTTP_401_UNAUTHORIZED: PROBLEM_RESPONSES[status.HTTP_401_UNAUTHORIZED],
+        status.HTTP_403_FORBIDDEN: PROBLEM_RESPONSES[status.HTTP_403_FORBIDDEN],
+        status.HTTP_400_BAD_REQUEST: PROBLEM_RESPONSES[status.HTTP_400_BAD_REQUEST],
+        status.HTTP_404_NOT_FOUND: PROBLEM_RESPONSES[status.HTTP_404_NOT_FOUND],
+        status.HTTP_422_UNPROCESSABLE_CONTENT: (
+            PROBLEM_RESPONSES[status.HTTP_422_UNPROCESSABLE_CONTENT]
+        ),
+        status.HTTP_500_INTERNAL_SERVER_ERROR: (
+            PROBLEM_RESPONSES[status.HTTP_500_INTERNAL_SERVER_ERROR]
+        ),
+    },
+)
+def on_gep_job_failed(
+    request: Request,
+    body: FailureEvent,
+    caller: Annotated[set[str], Depends(verify_callback_token)],
+    db: Session = Depends(get_db),
+) -> None:
+    """URIs for failed callbacks for jobs."""
+    _handle_callback_status_update(
+        db, body, ProcessingStatusEnum.FAILED, request, caller
+    )
diff --git a/app/platforms/implementations/gep/schemas.py b/app/platforms/implementations/gep/schemas.py
new file mode 100644
index 0000000..b4d84cb
--- /dev/null
+++ b/app/platforms/implementations/gep/schemas.py
@@ -0,0 +1,24 @@
+"""Payload models accepted by the GEP callback integration."""
+
+from pydantic import Field
+
+from app.config.schemas import BackendAuthConfig
+
+from stage_events_client.models import (
+    CompletedCloudEvent, DismissedCloudEvent, FailedCloudEvent,
+    PipedCloudEvent, PreparedCloudEvent, StagedCloudEvent, SubmittedCloudEvent,
+)
+
+ProgressEvent = (
+    SubmittedCloudEvent | DismissedCloudEvent | PreparedCloudEvent | StagedCloudEvent
+)
+SuccessEvent = PipedCloudEvent | CompletedCloudEvent
+FailureEvent = FailedCloudEvent
+
+
+class CallbackBackendAuthConfig(BackendAuthConfig):
+    """Backend authentication extended locally for inbound GEP callbacks."""
+
+    callback_issuer: str | None = None
+    callback_audience: str | None = None
+    callback_clients: list[str] = Field(default_factory=list)
diff --git a/app/platforms/implementations/gep/settings.py b/app/platforms/implementations/gep/settings.py
new file mode 100644
index 0000000..1bf2ee1
--- /dev/null
+++ b/app/platforms/implementations/gep/settings.py
@@ -0,0 +1,19 @@
+"""Read GEP callback settings without changing shared backend configuration."""
+
+from functools import lru_cache
+
+from pydantic import TypeAdapter
+
+from app.config.settings import settings
+from app.platforms.implementations.gep.schemas import CallbackBackendAuthConfig
+
+
+@lru_cache(maxsize=1)
+def _parse_callback_backends(raw: str) -> dict[str, CallbackBackendAuthConfig]:
+    return TypeAdapter(dict[str, CallbackBackendAuthConfig]).validate_json(raw)
+
+
+def get_callback_backends() -> dict[str, CallbackBackendAuthConfig]:
+    # The shared BackendAuthConfig ignores callback fields. Reparse the same raw
+    # BACKENDS value with our local extension, preserving the deployment contract.
+    return _parse_callback_backends(settings.backends or "{}")
diff --git a/app/platforms/implementations/gep/subscriber.py b/app/platforms/implementations/gep/subscriber.py
new file mode 100644
index 0000000..2b37269
--- /dev/null
+++ b/app/platforms/implementations/gep/subscriber.py
@@ -0,0 +1,12 @@
+"""GEP callback URLs hosted by the existing APEx Dispatcher."""
+
+CALLBACK_PREFIX = "/integrations/ogc-api-processes/gep/callbacks"
+
+
+def build_subscriber(base_url: str) -> dict[str, str]:
+    prefix = f"{base_url.rstrip('/')}{CALLBACK_PREFIX}"
+    return {
+        "successUri": f"{prefix}/success",
+        "inProgressUri": f"{prefix}/inProgress",
+        "failedUri": f"{prefix}/failed",
+    }
diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py
index f9cca3d..2d4039d 100644
--- a/app/platforms/implementations/ogc_api_process.py
+++ b/app/platforms/implementations/ogc_api_process.py
@@ -4,6 +4,8 @@
 
 import requests
 from app.auth import exchange_token, get_current_user_claims
+from app.config.settings import settings
+from app.platforms.implementations.gep.subscriber import build_subscriber
 from fastapi import Response
 from loguru import logger
 
@@ -50,7 +52,7 @@ class OGCAPIProcessPlatform(BaseProcessingPlatform):
     }
 
     status_mapping = {
-        StatusCode.ACCEPTED: ProcessingStatusEnum.CREATED,
+        StatusCode.ACCEPTED: ProcessingStatusEnum.QUEUED,
         StatusCode.RUNNING: ProcessingStatusEnum.RUNNING,
         StatusCode.DISMISSED: ProcessingStatusEnum.CANCELED,
         StatusCode.SUCCESSFUL: ProcessingStatusEnum.FINISHED,
@@ -210,10 +212,8 @@ async def execute_job(
         if user_claims.get("email"):
             properties["email"] = user_claims["email"]
 
-        data = {
-            "inputs": parameters,
-            "properties": properties,
-        }
+        data = {"inputs": parameters, "properties": properties}
+        data["subscriber"] = build_subscriber(settings.dispatch_api_public_url)
 
         content = api_client.execute_simple(
             process_id=details.application, execute=data, _headers=headers
diff --git a/app/routers/unit_jobs.py b/app/routers/unit_jobs.py
index 5770746..04395ae 100644
--- a/app/routers/unit_jobs.py
+++ b/app/routers/unit_jobs.py
@@ -25,7 +25,6 @@
     get_processing_job_by_user_id,
     get_processing_job_results,
 )
-
 from stac_pydantic import Collection
 
 # from app.auth import get_current_user
diff --git a/app/services/job_status.py b/app/services/job_status.py
new file mode 100644
index 0000000..a49932c
--- /dev/null
+++ b/app/services/job_status.py
@@ -0,0 +1,28 @@
+"""Platform-independent status updates for already-resolved jobs."""
+
+from sqlalchemy.orm import Session
+
+from app.database.models.processing_job import (
+    ProcessingJobRecord, update_job_status_by_id,
+)
+from app.schemas.enum import ProcessingStatusEnum
+from app.services.processing_state_machine import ProcessingStateMachine
+
+
+class InvalidJobTransition(ValueError):
+    """The requested status would violate the job lifecycle."""
+
+
+def apply_job_status(
+    database: Session,
+    job: ProcessingJobRecord,
+    target_status: ProcessingStatusEnum,
+) -> None:
+    if not ProcessingStateMachine.can_transition(job.status, target_status):
+        raise InvalidJobTransition(
+            f"Job {job.platform_job_id} can not transition from status "
+            f"{job.status} to {target_status}"
+        )
+    if job.status != target_status:
+        update_job_status_by_id(database, job.id, target_status)
+        job.status = target_status
diff --git a/app/services/processing.py b/app/services/processing.py
index 39b4f87..15ed59c 100644
--- a/app/services/processing.py
+++ b/app/services/processing.py
@@ -6,6 +6,7 @@
 from sqlalchemy.orm import Session
 from stac_pydantic import Collection
 
+
 from app.auth import get_current_user_id
 from app.database.models.processing_job import (
     ProcessingJobRecord,
@@ -57,7 +58,7 @@ async def create_processing_job(
             title=request.title,
             label=request.label,
             status=(
-                ProcessingStatusEnum.CREATED if job_id else ProcessingStatusEnum.FAILED
+                ProcessingStatusEnum.QUEUED if job_id else ProcessingStatusEnum.FAILED
             ),
             user_id=user,
             platform_job_id=job_id,
diff --git a/app/services/processing_state_machine.py b/app/services/processing_state_machine.py
new file mode 100644
index 0000000..8d295f1
--- /dev/null
+++ b/app/services/processing_state_machine.py
@@ -0,0 +1,89 @@
+"""Pure status validation; imports the existing APEx enum.
+
+No database writes or remote requests are performed. Confirmation flags must
+come from trusted integration code, never directly from a callback payload.
+"""
+
+from statemachine import State, StateMachine
+
+from app.schemas.enum import ProcessingStatusEnum as Status
+
+
+class ProcessingStateMachine(StateMachine):
+    """Execution observations plus explicitly confirmed resource deletion.
+
+    UNKNOWN describes missing knowledge, so it is handled by can_transition
+    rather than represented as an execution state. FINISHED/CANCELED/FAILED
+    and DELETED are final and have no outgoing transitions.
+    """
+
+    created = State(value=Status.CREATED, initial=True)
+    queued = State(value=Status.QUEUED)
+    running = State(value=Status.RUNNING)
+    finished = State(value=Status.FINISHED, final=True)
+    canceled = State(value=Status.CANCELED, final=True)
+    failed = State(value=Status.FAILED, final=True)
+    deleted = State(value=Status.DELETED, final=True)
+
+    observe_queued = created.to(queued)
+    observe_running = created.to(running) | queued.to(running)
+    observe_finished = (
+        created.to(finished) | queued.to(finished) | running.to(finished)
+    )
+    observe_canceled = (
+        created.to(canceled) | queued.to(canceled) | running.to(canceled)
+    )
+    observe_failed = created.to(failed) | queued.to(failed) | running.to(failed)
+    observe_deleted = (
+        created.to(deleted, cond="is_deletion_confirmed")
+        | queued.to(deleted, cond="is_deletion_confirmed")
+        | running.to(deleted, cond="is_deletion_confirmed")
+    )
+
+    @staticmethod
+    def is_deletion_confirmed(deletion_confirmed: bool = False) -> bool:
+        return deletion_confirmed
+
+    @classmethod
+    def can_transition(
+        cls,
+        current_status: Status | str,
+        new_status: Status | str,
+        *,
+        authoritative: bool = False,
+        deletion_confirmed: bool = False,
+    ) -> bool:
+        """Check an update without changing a machine, model, or database.
+
+        Accepts enum members or their serialized lowercase values. Invalid
+        values raise ValueError rather than being silently treated as UNKNOWN.
+
+        True means the update is allowed, including an idempotent no-op.
+        False means preserve current_status. Incoming UNKNOWN is never allowed
+        to replace a known status. Resolving a stored UNKNOWN requires a
+        successful authoritative lookup for the same execution attempt.
+
+        authoritative does not override known-state transition rules.
+        deletion_confirmed is required to enter DELETED, even from UNKNOWN.
+        Neither flag performs verification itself.
+        """
+        current = Status(current_status)
+        target = Status(new_status)
+
+        if current == target:
+            return True
+        if target == Status.UNKNOWN:
+            return False
+        if target == Status.DELETED and not deletion_confirmed:
+            return False
+        if current == Status.UNKNOWN:
+            return authoritative
+
+        machine = cls(start_value=current)
+        event_name = f"observe_{target.value}"
+        return any(
+            event.id == event_name
+            for event in machine.enabled_events(
+                deletion_confirmed=deletion_confirmed,
+            )
+        )
diff --git a/docs/configuration.md b/docs/configuration.md
index 70cb4e2..05b7225 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -67,4 +67,41 @@ KEYCLOAK_CLIENT_SECRET=apex-client-secret
 
 
 BACKENDS='{"https://openeo.backend1.com" {"auth_method": "CLIENT_CREDENTIALS", "client_credentials": "oidc_provider/client_id/secret_secret"}, "https://openeo.backend2.com" {"auth_method": "USER_CREDENTIALS",  "token_provider": "backend", "token_prefix": "oidc/backend"}}'
-```
\ No newline at end of file
+```
+
+## GEP callbacks for OGC API Processes
+
+GEP callback routes run inside the existing APEx Dispatcher. OGC job execution
+continues to attach a subscriber block automatically, using the existing
+`DISPATCH_API_PUBLIC_URL` setting. No additional backend configuration is required.
+
+The callback URLs are rooted at `DISPATCH_API_PUBLIC_URL`:
+
+- `/integrations/ogc-api-processes/gep/callbacks/inProgress`
+- `/integrations/ogc-api-processes/gep/callbacks/success`
+- `/integrations/ogc-api-processes/gep/callbacks/failed`
+
+These endpoints accept GEP stage CloudEvents, not arbitrary OGC API Processes
+notification payloads. Routes are registered explicitly in `app/main.py`;
+`app/platforms/implementations/gep/__init__.py` has no registration side effects.
+
+The GEP adapter owns event mapping, subject normalization and subscriber URLs.
+It resolves exact platform job identifiers for OGC jobs;
+it does not accept APEx database IDs or suffix matches. Multiple matching jobs
+are rejected rather than arbitrarily selecting a backend. The existing subject
+format does not identify a trusted backend; backend authentication/correlation
+requires a separate change. This refactor does not implement callback authentication.
+
+The shared `app/services/job_status.py` validates and persists status changes.
+Duplicate statuses are no-ops; the existing state machine rejects terminal-state
+regressions. Existing polling behavior is unchanged. Missing jobs on progress
+callbacks still receive 200, preserving the existing early-callback behavior;
+missing jobs on success/failure receive 404.
+
+### Migration
+
+This change removes the three `/unit_jobs/{inProgress,success,failed}` callback
+routes. Finish jobs
+already subscribed to old URLs before deployment, or provide temporary explicit
+reverse-proxy rewrites to the new URLs until those jobs complete. No database
+migration or new `ProcessTypeEnum` value is required.
diff --git a/env.example b/env.example
index 7d0762e..15a036d 100644
--- a/env.example
+++ b/env.example
@@ -6,6 +6,7 @@ APP_ENV=
 
 # CORS
 CORS_ALLOWED_ORIGINS=
+DISPATCH_API_PUBLIC_URL=
 
 # Keycloak
 KEYCLOAK_HOST=
diff --git a/openapi.json b/openapi.json
index eafe5d0..e3b0580 100644
--- a/openapi.json
+++ b/openapi.json
@@ -711,6 +711,108 @@
           }
         }
       }
+    },
+    "/integrations/ogc-api-processes/gep/callbacks/success": {
+      "post": {
+        "operationId": "onGepJobSuccess",
+        "tags": ["GEP callbacks"],
+        "summary": "URIs for success callbacks for jobs.",
+        "description": "Accepts GEP stage CloudEvents. The callback URL will be invoked after a job succeeded its execution.\n",
+        "requestBody": {
+          "$ref": "#/components/requestBodies/SuccessRequestBody"
+        },
+        "responses": {
+          "200": {
+            "$ref": "#/components/responses/Ok"
+          },
+          "400": {
+            "$ref": "#/components/responses/BadRequest"
+          },
+          "401": {
+            "$ref": "#/components/responses/Unauthorized"
+          },
+          "403": {
+            "$ref": "#/components/responses/Forbidden"
+          },
+          "404": {
+            "$ref": "#/components/responses/NotFound"
+          },
+          "422": {
+            "$ref": "#/components/responses/BusinessRuleViolation"
+          },
+          "500": {
+            "$ref": "#/components/responses/ServerError"
+          }
+        }
+      }
+    },
+    "/integrations/ogc-api-processes/gep/callbacks/inProgress": {
+      "post": {
+        "operationId": "onGepJobProgress",
+        "tags": ["GEP callbacks"],
+        "summary": "URIs for progress callbacks for jobs.",
+        "description": "Accepts GEP stage CloudEvents. The callback URL will be invoked while a job is changing its status during its execution.\n",
+        "requestBody": {
+          "$ref": "#/components/requestBodies/InProgressRequestBody"
+        },
+        "responses": {
+          "200": {
+            "$ref": "#/components/responses/Ok"
+          },
+          "400": {
+            "$ref": "#/components/responses/BadRequest"
+          },
+          "401": {
+            "$ref": "#/components/responses/Unauthorized"
+          },
+          "403": {
+            "$ref": "#/components/responses/Forbidden"
+          },
+          "404": {
+            "$ref": "#/components/responses/NotFound"
+          },
+          "422": {
+            "$ref": "#/components/responses/BusinessRuleViolation"
+          },
+          "500": {
+            "$ref": "#/components/responses/ServerError"
+          }
+        }
+      }
+    },
+    "/integrations/ogc-api-processes/gep/callbacks/failed": {
+      "post": {
+        "operationId": "onGepJobFailed",
+        "tags": ["GEP callbacks"],
+        "summary": "URIs for failed callbacks for jobs.",
+        "description": "Accepts GEP stage CloudEvents. The callback URL will be invoked after a job failed its execution.\n",
+        "requestBody": {
+          "$ref": "#/components/requestBodies/FailedRequestBody"
+        },
+        "responses": {
+          "200": {
+            "$ref": "#/components/responses/Ok"
+          },
+          "400": {
+            "$ref": "#/components/responses/BadRequest"
+          },
+          "401": {
+            "$ref": "#/components/responses/Unauthorized"
+          },
+          "403": {
+            "$ref": "#/components/responses/Forbidden"
+          },
+          "404": {
+            "$ref": "#/components/responses/NotFound"
+          },
+          "422": {
+            "$ref": "#/components/responses/BusinessRuleViolation"
+          },
+          "500": {
+            "$ref": "#/components/responses/ServerError"
+          }
+        }
+      }
     }
   },
   "components": {
@@ -2545,6 +2647,95 @@
           }
         }
       }
+    },
+    "parameters": {
+      "job_id": {
+        "name": "job_id",
+        "in": "path",
+        "description": "local identifier of a Job",
+        "required": true,
+        "schema": {
+          "type": "string"
+        }
+      }
+    },
+    "requestBodies": {
+      "SuccessRequestBody": {
+        "description": "A structured-mode CloudEvent request body for successes.",
+        "required": true,
+        "content": {
+          "application/cloudevents+json": {
+            "schema": {
+              "oneOf": [
+                {
+                  "$ref": "https://terradue.github.io/stage-events-client/c4/class/schemas/openapi.yaml#/components/schemas/PipedCloudEvent"
+                },
+                {
+                  "$ref": "https://terradue.github.io/stage-events-client/c4/class/schemas/openapi.yaml#/components/schemas/CompletedCloudEvent"
+                }
+              ]
+            }
+          }
+        }
+      },
+      "InProgressRequestBody": {
+        "description": "A structured-mode CloudEvent request body for in-progress updates.",
+        "required": true,
+        "content": {
+          "application/cloudevents+json": {
+            "schema": {
+              "oneOf": [
+                {
+                  "$ref": "https://terradue.github.io/stage-events-client/c4/class/schemas/openapi.yaml#/components/schemas/SubmittedCloudEvent"
+                },
+                {
+                  "$ref": "https://terradue.github.io/stage-events-client/c4/class/schemas/openapi.yaml#/components/schemas/DismissedCloudEvent"
+                },
+                {
+                  "$ref": "https://terradue.github.io/stage-events-client/c4/class/schemas/openapi.yaml#/components/schemas/PreparedCloudEvent"
+                },
+                {
+                  "$ref": "https://terradue.github.io/stage-events-client/c4/class/schemas/openapi.yaml#/components/schemas/StagedCloudEvent"
+                }
+              ]
+            }
+          }
+        }
+      },
+      "FailedRequestBody": {
+        "description": "A structured-mode CloudEvent request body for failures.",
+        "required": true,
+        "content": {
+          "application/cloudevents+json": {
+            "schema": {
+              "$ref": "https://terradue.github.io/stage-events-client/c4/class/schemas/openapi.yaml#/components/schemas/FailedCloudEvent"
+            }
+          }
+        }
+      }
+    },
+    "responses": {
+      "Ok": {
+        "description": "Request has been processed and job status has been successfully updated.\n"
+      },
+      "BadRequest": {
+        "$ref": "https://terradue.github.io/http-apis-problems-detail/openapi.yaml#/components/responses/BadRequest"
+      },
+      "BusinessRuleViolation": {
+        "$ref": "https://terradue.github.io/http-apis-problems-detail/openapi.yaml#/components/responses/BusinessRuleViolation"
+      },
+      "Unauthorized": {
+        "$ref": "https://terradue.github.io/http-apis-problems-detail/openapi.yaml#/components/responses/Unauthorized"
+      },
+      "Forbidden": {
+        "$ref": "https://terradue.github.io/http-apis-problems-detail/openapi.yaml#/components/responses/Forbidden"
+      },
+      "NotFound": {
+        "$ref": "https://terradue.github.io/http-apis-problems-detail/openapi.yaml#/components/responses/NotFound"
+      },
+      "ServerError": {
+        "$ref": "https://terradue.github.io/http-apis-problems-detail/openapi.yaml#/components/responses/ServerError"
+      }
     }
   }
-}
\ No newline at end of file
+}
diff --git a/requirements.txt b/requirements.txt
index ef520c2..1056dd8 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -32,3 +32,6 @@ types-requests
 types-shapely
 uvicorn[standard]
 ogc-api-processes-client==0.6.0
+http-apis-problems-detail>=2
+stage-events-client==1.3.0
+python-statemachine==3.2.2
diff --git a/tests/platforms/integrations/gep/test_gep_auth.py b/tests/platforms/integrations/gep/test_gep_auth.py
new file mode 100644
index 0000000..fbfb013
--- /dev/null
+++ b/tests/platforms/integrations/gep/test_gep_auth.py
@@ -0,0 +1,169 @@
+"""Offline security tests; all GEP feature changes remain inside this package."""
+import json
+import time
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives.asymmetric import rsa
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from http_apis_problems_detail.fastapi import (
+    ProblemRegistryException, problem_registry_exception_handler,
+)
+
+from app.config.settings import settings
+from app.platforms.implementations.gep import auth
+from app.platforms.implementations.gep import router as routes
+from app.platforms.implementations.gep.settings import get_callback_backends
+from app.platforms.implementations.gep.subscriber import CALLBACK_PREFIX
+from app.schemas.enum import ProcessingStatusEnum
+
+ISSUER = "https://iam.example.test/realms/callbacks"
+ENDPOINT = "https://mate.example.test"
+
+
+@pytest.fixture
+def harness(monkeypatch):
+    backend = dict(token_provider="terradue", callback_issuer=ISSUER,
+                   callback_audience="apex-dispatch-api",
+                   callback_clients=["mate-api-callbacks"])
+    monkeypatch.setattr(settings, "backends", json.dumps({ENDPOINT: backend}))
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    jwks = MagicMock()
+    jwks.get_signing_key_from_jwt.return_value.key = key.public_key()
+    factory = MagicMock(return_value=jwks)
+    monkeypatch.setattr(auth, "_callback_jwks_client", factory)
+    job = SimpleNamespace(id=1, status=ProcessingStatusEnum.RUNNING,
+                          service=json.dumps({"endpoint": ENDPOINT}))
+    lookup = MagicMock(return_value=job)
+    update = MagicMock()
+    monkeypatch.setattr(routes, "get_gep_job", lookup)
+    monkeypatch.setattr(routes, "apply_job_status", update)
+    api = FastAPI()
+    api.include_router(routes.router)
+    api.add_exception_handler(ProblemRegistryException, problem_registry_exception_handler)
+    api.dependency_overrides[routes.get_db] = lambda: MagicMock()
+
+    def token(**changes):
+        claims = dict(iss=ISSUER, aud="apex-dispatch-api", azp="mate-api-callbacks",
+                      exp=int(time.time()) + 300)
+        claims.update(changes)
+        return jwt.encode({k: v for k, v in claims.items() if v is not None},
+                          key, algorithm="RS256", headers={"kid": "test"})
+
+    with TestClient(api) as client:
+        yield SimpleNamespace(client=client, token=token, lookup=lookup, update=update,
+                              job=job, factory=factory, backend=backend, api=api)
+
+
+def post(h, endpoint="success", token=None):
+    event_type = {"success": "piped", "inProgress": "staged", "failed": "failed"}[endpoint]
+    payload = dict(source="terradue:apex:test", subject="default:uid:job-123",
+                   partitionkey="part-1", type=event_type,
+                   data=dict(namespace="default", process_id="test-proc", job_id="job-123",
+                             entries={}, outputs={}, usage={}))
+    return h.client.post(CALLBACK_PREFIX + "/" + endpoint, json=payload,
+                         headers={} if token is None else {"Authorization": token})
+
+
+@pytest.mark.parametrize("endpoint", ["success", "inProgress", "failed"])
+@pytest.mark.parametrize("scenario,expected", [
+    ("valid", 200), ("missing", 401), ("basic", 401), ("malformed", 401),
+    ("expired", 401), ("issuer", 401), ("audience", 401), ("no_exp", 401),
+    ("no_aud", 401), ("no_iss", 401), ("future_nbf", 401), ("future_iat", 401),
+    ("client", 403), ("no_azp", 403), ("other_job", 403), ("slash", 200),
+    ("bad_signature", 401), ("jwks_failure", 401), ("algorithm", 401),
+])
+def test_callback_security(harness, endpoint, scenario, expected):
+    h = harness
+    changes = {"expired": {"exp": 1}, "issuer": {"iss": "https://unknown.test"},
+               "audience": {"aud": "other"}, "no_exp": {"exp": None},
+               "no_aud": {"aud": None}, "no_iss": {"iss": None},
+               "future_nbf": {"nbf": 9999999999}, "future_iat": {"iat": 9999999999},
+               "client": {"azp": "other"}, "no_azp": {"azp": None}}
+    token = "Bearer " + h.token(**changes.get(scenario, {}))
+    if scenario == "missing":
+        token = None
+    elif scenario == "basic":
+        token = "Basic abc"
+    elif scenario == "malformed":
+        token = "Bearer invalid"
+    elif scenario in ("other_job", "slash"):
+        target = "https://other.test" if scenario == "other_job" else ENDPOINT + "/"
+        h.job.service = json.dumps({"endpoint": target})
+    elif scenario == "bad_signature":
+        h.factory.return_value.get_signing_key_from_jwt.return_value.key = rsa.generate_private_key(
+            public_exponent=65537, key_size=2048).public_key()
+    elif scenario == "jwks_failure":
+        h.factory.return_value.get_signing_key_from_jwt.side_effect = (
+            jwt.PyJWKClientConnectionError("offline")
+        )
+    elif scenario == "algorithm":
+        token = "Bearer " + jwt.encode({"iss": ISSUER}, "x" * 32, algorithm="HS256")
+    response = post(h, endpoint, token)
+    assert response.status_code == expected, response.text
+    if expected == 200:
+        h.update.assert_called_once()
+    else:
+        assert response.headers["content-type"].startswith("application/problem+json")
+        assert response.json()["status"] == expected
+        assert response.json()["instance"].endswith(CALLBACK_PREFIX + "/" + endpoint)
+        h.update.assert_not_called()
+    if expected == 401:
+        assert response.headers["www-authenticate"] == "Bearer"
+        h.lookup.assert_not_called()
+    if scenario in ("issuer", "no_iss"):
+        h.factory.assert_not_called()
+
+
+@pytest.mark.parametrize("other_policy", ["client", "audience", "disabled"])
+def test_shared_issuer_isolation(harness, monkeypatch, other_policy):
+    h = harness
+    other = dict(h.backend)
+    other.update({"client": {"callback_clients": ["other"]},
+                  "audience": {"callback_audience": "other"},
+                  "disabled": {"callback_clients": []}}[other_policy])
+    backends_config = {ENDPOINT: h.backend, "https://other.test": other}
+    monkeypatch.setattr(settings, "backends", json.dumps(backends_config))
+    h.job.service = json.dumps({"endpoint": "https://other.test"})
+    assert post(h, token="Bearer " + h.token()).status_code == 403
+    h.update.assert_not_called()
+
+
+def test_raw_backends_retains_local_fields(harness):
+    # Shared config deliberately remains unchanged and drops these extra fields.
+    from app.config.schemas import BackendAuthConfig
+    assert not hasattr(BackendAuthConfig(**harness.backend), "callback_issuer")
+    assert get_callback_backends()[ENDPOINT].callback_issuer == ISSUER
+
+
+@pytest.mark.parametrize("raw", ["", "{}"])
+def test_no_config_fails_closed(harness, monkeypatch, raw):
+    monkeypatch.setattr(settings, "backends", raw)
+    assert post(harness, token="Bearer " + harness.token()).status_code == 401
+    harness.factory.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "endpoint,expected", [("success", 404), ("failed", 404), ("inProgress", 200)]
+)
+def test_unknown_job(harness, endpoint, expected):
+    harness.lookup.return_value = None
+    assert post(harness, endpoint, "Bearer " + harness.token()).status_code == expected
+    harness.update.assert_not_called()
+
+
+def test_transition_failure(harness):
+    harness.update.side_effect = routes.InvalidJobTransition("Invalid transition")
+    assert post(harness, token="Bearer " + harness.token()).status_code == 422
+
+
+def test_openapi(harness):
+    schema = harness.api.openapi()
+    for endpoint in ("success", "inProgress", "failed"):
+        operation = schema["paths"][CALLBACK_PREFIX + "/" + endpoint]["post"]
+        assert {"GepCallbackBearer": []} in operation["security"]
+        for code in ("401", "403"):
+            assert "application/problem+json" in operation["responses"][code]["content"]
diff --git a/tests/platforms/integrations/gep/test_integration.py b/tests/platforms/integrations/gep/test_integration.py
new file mode 100644
index 0000000..e3cd533
--- /dev/null
+++ b/tests/platforms/integrations/gep/test_integration.py
@@ -0,0 +1,97 @@
+"""GEP job correlation, subscriber URLs, and shared lifecycle regressions."""
+
+import json
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import Session
+
+from app.database.models.processing_job import ProcessingJobRecord
+from app.platforms.implementations.gep.callbacks import get_gep_job
+from app.platforms.implementations.gep.subscriber import (
+    CALLBACK_PREFIX, build_subscriber,
+)
+from app.schemas.enum import ProcessingStatusEnum as Status
+from app.services.job_status import apply_job_status, InvalidJobTransition
+
+
+def test_subscriber_uses_dispatcher_url():
+    expected = {
+        "successUri": "https://dispatch.example" + CALLBACK_PREFIX + "/success",
+        "inProgressUri": "https://dispatch.example" + CALLBACK_PREFIX + "/inProgress",
+        "failedUri": "https://dispatch.example" + CALLBACK_PREFIX + "/failed",
+    }
+    assert build_subscriber('https://dispatch.example/') == expected
+    assert build_subscriber('https://dispatch.example') == expected
+
+
+@pytest.fixture
+def jobs_database():
+    engine = create_engine('sqlite:///:memory:')
+    # Portable fixture for the MySQL LONGTEXT-backed production ORM model.
+    with engine.begin() as conn:
+        conn.execute(text('''CREATE TABLE processing_jobs (
+            id INTEGER PRIMARY KEY, title TEXT, label TEXT, status TEXT,
+            user_id TEXT, platform_job_id TEXT, parameters TEXT, service TEXT,
+            created DATETIME, updated DATETIME, upscaling_task_id INTEGER
+        )'''))
+        rows = [
+            (1, 'OPENEO', 'ns:job', 'https://gep.example'),
+            (2, 'OGC_API_PROCESS', 'ns:other-job', 'https://other.example'),
+            (3, 'OGC_API_PROCESS', 'other:job', 'https://gep.example'),
+            (4, 'OGC_API_PROCESS', 'ns:job', 'https://gep.example'),
+            (5, 'OGC_API_PROCESS', '123', 'https://gep.example'),
+        ]
+        for job_id, label, platform_id, endpoint in rows:
+            conn.execute(text('''INSERT INTO processing_jobs
+                (id, label, status, platform_job_id, service)
+                VALUES (:id, :label, 'RUNNING', :platform_id, :service)'''),
+                         dict(id=job_id, label=label, platform_id=platform_id,
+                              service=json.dumps(dict(endpoint=endpoint, application='app'))))
+    with Session(engine) as session:
+        yield session
+    engine.dispose()
+
+
+def test_lookup_is_exact_and_scoped_to_ogc(jobs_database):
+    assert get_gep_job(jobs_database, 'ns:job').id == 4
+    assert get_gep_job(jobs_database, 'ns:other-job').id == 2
+    assert get_gep_job(jobs_database, 'job') is None
+    assert get_gep_job(jobs_database, '4') is None
+    assert get_gep_job(jobs_database, '123').id == 5
+
+
+def test_lookup_rejects_ambiguity(jobs_database):
+    jobs_database.execute(text('''INSERT INTO processing_jobs
+        (id, label, status, platform_job_id, service)
+        SELECT 6, label, status, platform_job_id, service
+        FROM processing_jobs WHERE id = 4'''))
+    with pytest.raises(ValueError, match='Ambiguous'):
+        get_gep_job(jobs_database, 'ns:job')
+
+
+@pytest.mark.parametrize('current', [Status.FINISHED, Status.FAILED, Status.CANCELED])
+def test_terminal_state_cannot_regress(current):
+    job = ProcessingJobRecord(id=1, status=current, platform_job_id='ns:job')
+    with patch('app.services.job_status.update_job_status_by_id') as update:
+        with pytest.raises(InvalidJobTransition):
+            apply_job_status(None, job, Status.RUNNING)
+        update.assert_not_called()
+        assert job.status == current
+
+
+def test_duplicate_status_is_noop():
+    job = ProcessingJobRecord(id=1, status=Status.FINISHED)
+    with patch('app.services.job_status.update_job_status_by_id') as update:
+        apply_job_status(None, job, Status.FINISHED)
+        update.assert_not_called()
+
+
+def test_routes_and_openapi(client):
+    paths = client.get('/openapi.json').json()['paths']
+    for event in ('success', 'inProgress', 'failed'):
+        assert CALLBACK_PREFIX + '/' + event in paths
+        assert '/unit_jobs/' + event not in paths
+        response = client.post('/unit_jobs/' + event, json={})
+        assert response.status_code in (404, 405)
diff --git a/tests/platforms/integrations/gep/test_router.py b/tests/platforms/integrations/gep/test_router.py
new file mode 100644
index 0000000..d8d201c
--- /dev/null
+++ b/tests/platforms/integrations/gep/test_router.py
@@ -0,0 +1,351 @@
+from unittest.mock import ANY, MagicMock, patch
+import pytest
+from fastapi import status
+
+from app.main import app
+from app.platforms.implementations.gep.auth import verify_callback_token
+from app.schemas.enum import OutputFormatEnum, ProcessingStatusEnum
+from app.schemas.unit_job import ServiceDetails
+from app.platforms.implementations.ogc_api_process import OGCAPIProcessPlatform
+
+
+# --- Test Payloads ---
+PIPED_PAYLOAD = {
+    "source": "terradue:apex:test",
+    "subject": "default:test-uid:job-123",
+    "partitionkey": "part-1",
+    "type": "piped",
+    "data": {
+        "namespace": "default",
+        "process_id": "test-proc",
+        "job_id": "job-123",
+        "entries": {},
+    },
+}
+
+STAGED_PAYLOAD = {
+    "source": "terradue:apex:test",
+    "subject": "default:test-uid:job-123",
+    "partitionkey": "part-1",
+    "type": "staged",
+    "data": {
+        "namespace": "default",
+        "process_id": "test-proc",
+        "job_id": "job-123",
+        "outputs": {},
+    },
+}
+
+PREPARED_PAYLOAD = {
+    "source": "terradue:apex:test",
+    "subject": "default:test-uid:job-123",
+    "partitionkey": "part-1",
+    "type": "prepared",
+    "data": {
+        "namespace": "default",
+        "process_id": "test-proc",
+        "job_id": "job-123",
+        "inputs": {},
+    },
+}
+
+SUBMITTED_PAYLOAD = {
+    "source": "terradue:apex:test",
+    "subject": "default:test-uid:job-123",
+    "partitionkey": "part-1",
+    "type": "submitted",
+    "data": {
+        "namespace": "default",
+        "process_id": "test-proc",
+    },
+}
+
+DISMISSED_PAYLOAD = {
+    "source": "terradue:apex:test",
+    "subject": "default:test-uid:job-123",
+    "partitionkey": "part-1",
+    "type": "dismissed",
+    "data": {
+        "namespace": "default",
+        "status": "Pending",
+    },
+}
+
+FAILED_PAYLOAD = {
+    "source": "terradue:apex:test",
+    "subject": "default:test-uid:job-123",
+    "partitionkey": "part-1",
+    "type": "failed",
+    "data": {
+        "namespace": "default",
+        "process_id": "test-proc",
+        "job_id": "job-123",
+        "usage": {},
+    },
+}
+
+COMPLETED_PAYLOAD = {
+    "source": "terradue:apex:test",
+    "subject": "default:test-uid:job-123",
+    "partitionkey": "part-1",
+    "type": "completed",
+    "data": {
+        "namespace": "default",
+        "process_id": "test-proc",
+        "job_id": "job-123",
+        "outputs": {},
+        "usage": {},
+    },
+}
+
+
+# --- Inbound Callback Tests ---
+
+@pytest.fixture
+def callback_dependencies(fake_processing_job_record):
+    fake_processing_job_record.platform_job_id = "default:job-123"
+    # Ensure authorized caller matches the endpoint in fake_processing_job_record.service ("foo")
+    app.dependency_overrides[verify_callback_token] = lambda: {"foo"}
+    with (
+        patch(
+            "app.platforms.implementations.gep.router.get_gep_job",
+            return_value=fake_processing_job_record,
+        ) as lookup,
+        patch(
+            "app.services.job_status.ProcessingStateMachine.can_transition",
+            return_value=True,
+        ) as can_transition,
+        patch("app.services.job_status.update_job_status_by_id") as update,
+    ):
+        yield lookup, can_transition, update
+    app.dependency_overrides.pop(verify_callback_token, None)
+
+
+CALLBACK_CASES = [
+    pytest.param("success", PIPED_PAYLOAD, ProcessingStatusEnum.FINISHED, id="piped"),
+    pytest.param("success", COMPLETED_PAYLOAD, ProcessingStatusEnum.FINISHED, id="completed"),
+    pytest.param("inProgress", STAGED_PAYLOAD, ProcessingStatusEnum.RUNNING, id="staged"),
+    pytest.param("inProgress", PREPARED_PAYLOAD, ProcessingStatusEnum.RUNNING, id="prepared"),
+    pytest.param("inProgress", SUBMITTED_PAYLOAD, ProcessingStatusEnum.QUEUED, id="submitted"),
+    pytest.param("inProgress", DISMISSED_PAYLOAD, ProcessingStatusEnum.CANCELED, id="dismissed"),
+    pytest.param("failed", FAILED_PAYLOAD, ProcessingStatusEnum.FAILED, id="failed"),
+]
+
+
+@pytest.mark.parametrize("endpoint,payload,target_status", CALLBACK_CASES)
+def test_callback_updates_job_status(
+    client, callback_dependencies, fake_processing_job_record,
+    endpoint, payload, target_status,
+):
+    lookup, can_transition, update = callback_dependencies
+    initial_status = fake_processing_job_record.status
+
+    response = client.post(
+        f"/integrations/ogc-api-processes/gep/callbacks/{endpoint}",
+        json=payload,
+        headers={"Content-Type": "application/cloudevents+json"},
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    lookup.assert_called_once_with(ANY, "default:job-123")
+    can_transition.assert_called_once_with(initial_status, target_status)
+    update.assert_called_once_with(
+        lookup.call_args.args[0], fake_processing_job_record.id, target_status
+    )
+    assert fake_processing_job_record.status == target_status
+
+
+@pytest.mark.parametrize("endpoint,payload,target_status", CALLBACK_CASES)
+def test_callback_rejects_invalid_transition(
+    client, callback_dependencies, fake_processing_job_record,
+    endpoint, payload, target_status,
+):
+    lookup, can_transition, update = callback_dependencies
+    fake_processing_job_record.status = ProcessingStatusEnum.DELETED
+    can_transition.return_value = False
+
+    response = client.post(
+        f"/integrations/ogc-api-processes/gep/callbacks/{endpoint}",
+        json=payload,
+        headers={"Content-Type": "application/cloudevents+json"},
+    )
+
+    assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
+    assert response.headers["content-type"] == "application/problem+json"
+    data = response.json()
+    assert data["status"] == 422
+    assert data["instance"].endswith(f"/integrations/ogc-api-processes/gep/callbacks/{endpoint}")
+    assert len(data["errors"]) == 1
+    assert data["errors"][0]["pointer"] == "subject"
+    assert "default:job-123" in data["errors"][0]["detail"]
+    lookup.assert_called_once_with(ANY, "default:job-123")
+    can_transition.assert_called_once_with(ProcessingStatusEnum.DELETED, target_status)
+    update.assert_not_called()
+    assert fake_processing_job_record.status == ProcessingStatusEnum.DELETED
+
+
+@pytest.mark.parametrize("endpoint,payload", [
+    ("success", PIPED_PAYLOAD),
+    ("failed", FAILED_PAYLOAD),
+])
+def test_callback_missing_job_404_problem_details(
+    client, callback_dependencies, endpoint, payload,
+):
+    lookup, can_transition, update = callback_dependencies
+    lookup.return_value = None
+
+    response = client.post(
+        f"/integrations/ogc-api-processes/gep/callbacks/{endpoint}",
+        json={**payload, "subject": "default:test-uid:nonexistent"},
+        headers={"Content-Type": "application/cloudevents+json"},
+    )
+
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.headers["content-type"] == "application/problem+json"
+    data = response.json()
+    assert data["status"] == 404
+    assert data["title"] == "Not Found"
+    assert data["instance"].endswith(f"/integrations/ogc-api-processes/gep/callbacks/{endpoint}")
+    assert data["errors"] == [
+        {
+            "detail": "Job default:nonexistent does not exist in the Job registry",
+            "pointer": "subject",
+        }
+    ]
+    lookup.assert_called_once_with(ANY, "default:nonexistent")
+    can_transition.assert_not_called()
+    update.assert_not_called()
+
+
+@pytest.mark.parametrize("payload", [
+    SUBMITTED_PAYLOAD, PREPARED_PAYLOAD, STAGED_PAYLOAD, DISMISSED_PAYLOAD,
+])
+def test_callback_in_progress_uncommitted_job_acknowledges_cleanly(
+    client, callback_dependencies, payload,
+):
+    lookup, can_transition, update = callback_dependencies
+    lookup.return_value = None
+
+    response = client.post(
+        "/integrations/ogc-api-processes/gep/callbacks/inProgress",
+        json=payload,
+        headers={"Content-Type": "application/cloudevents+json"},
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    lookup.assert_called_once_with(ANY, "default:job-123")
+    can_transition.assert_not_called()
+    update.assert_not_called()
+
+
+@pytest.mark.parametrize("failure_point", ["lookup", "transition", "update"])
+def test_callback_unexpected_error_returns_500(
+    client, callback_dependencies, fake_processing_job_record, failure_point,
+):
+    lookup, can_transition, update = callback_dependencies
+    initial_status = fake_processing_job_record.status
+    dependencies = {"lookup": lookup, "transition": can_transition, "update": update}
+    dependencies[failure_point].side_effect = RuntimeError("callback failure")
+
+    response = client.post(
+        "/integrations/ogc-api-processes/gep/callbacks/success",
+        json=PIPED_PAYLOAD,
+        headers={"Content-Type": "application/cloudevents+json"},
+    )
+
+    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+    assert response.headers["content-type"] == "application/problem+json"
+    data = response.json()
+    assert data["status"] == 500
+    assert data["instance"].endswith("/integrations/ogc-api-processes/gep/callbacks/success")
+    assert data["errors"] == [{
+        "parameter": "job_id",
+        "detail": (
+            "An error occurred when updating status of Job default:job-123: "
+            "callback failure"
+        ),
+    }]
+    assert fake_processing_job_record.status == initial_status
+    if failure_point != "update":
+        update.assert_not_called()
+    if failure_point == "lookup":
+        can_transition.assert_not_called()
+
+
+def testextract_job_identifier():
+    from app.platforms.implementations.gep.callbacks import extract_job_identifier
+
+    # 3-part subject (<namespace>:<uid>:<workflow_name>)
+    event = MagicMock(subject="myns:uid-1234:wf-5678")
+    assert extract_job_identifier(event) == "myns:wf-5678"
+
+    # 2-part subject
+    event = MagicMock(subject="myns:wf-5678")
+    assert extract_job_identifier(event) == "myns:wf-5678"
+
+    # Single string
+    event = MagicMock(subject="wf-5678")
+    assert extract_job_identifier(event) == "wf-5678"
+
+    # Missing / None subject
+    event = MagicMock(subject=None)
+    assert extract_job_identifier(event) is None
+
+
+# --- Outbound Subscriber Payload Test ---
+
+@pytest.mark.asyncio
+@patch("app.platforms.implementations.ogc_api_process.exchange_token")
+@patch("app.platforms.implementations.ogc_api_process.get_current_user_claims")
+async def test_ogc_platform_execute_job_injects_subscriber(
+    mock_claims, mock_exchange, monkeypatch
+):
+    mock_exchange.return_value = "fake_exchanged_token"
+    mock_claims.return_value = {
+        "sub": "user-sub",
+        "preferred_username": "testuser",
+        "email": "test@example.com",
+    }
+
+    platform = OGCAPIProcessPlatform()
+
+    mock_client = MagicMock()
+    mock_content = MagicMock()
+    mock_content.job_id = "ogc-exec-999"
+    mock_client.execute_simple.return_value = mock_content
+
+    with patch.object(
+        platform, "_create_api_client_instance", return_value=mock_client
+    ), patch.object(
+        platform, "_transform_parameters", return_value={"param1": "val1"}
+    ):
+        details = ServiceDetails(
+            endpoint="https://ades.example.com",
+            application="my-app",
+            namespace="my-ns",
+        )
+        job_id = await platform.execute_job(
+            user_token="user_token_123",
+            title="My Execution",
+            details=details,
+            parameters={"param1": "val1"},
+            format=OutputFormatEnum.JSON,
+        )
+
+        assert job_id == "my-ns:ogc-exec-999"
+
+        # Verify execute_simple was called with the subscriber block
+        mock_client.execute_simple.assert_called_once()
+        _, kwargs = mock_client.execute_simple.call_args
+        execute_data = kwargs["execute"]
+        assert "subscriber" in execute_data
+        subscriber = execute_data["subscriber"]
+        assert subscriber["successUri"].endswith(
+            "/integrations/ogc-api-processes/gep/callbacks/success"
+        )
+        assert subscriber["inProgressUri"].endswith(
+            "/integrations/ogc-api-processes/gep/callbacks/inProgress"
+        )
+        assert subscriber["failedUri"].endswith(
+            "/integrations/ogc-api-processes/gep/callbacks/failed"
+        )
diff --git a/tests/platforms/test_ogc_api_process_platform.py b/tests/platforms/test_ogc_api_process_platform.py
index 9d9cbea..e33c1eb 100644
--- a/tests/platforms/test_ogc_api_process_platform.py
+++ b/tests/platforms/test_ogc_api_process_platform.py
@@ -353,7 +353,7 @@ def test_split_job_id(platform, job_id, expected):
 @pytest.mark.parametrize(
     ("ogc_status, expected_status"),
     [
-        ("accepted", ProcessingStatusEnum.CREATED),
+        ("accepted", ProcessingStatusEnum.QUEUED),
         ("running", ProcessingStatusEnum.RUNNING),
         ("dismissed", ProcessingStatusEnum.CANCELED),
         ("successful", ProcessingStatusEnum.FINISHED),
@@ -478,6 +478,17 @@ async def test_execute_job_returns_namespaced_job_id(
                 "username": "alice",
                 "email": "alice@example.com",
             },
+            "subscriber": {
+                "successUri": (
+                    "http://localhost:8000/integrations/ogc-api-processes/gep/callbacks/success"
+                ),
+                "inProgressUri": (
+                    "http://localhost:8000/integrations/ogc-api-processes/gep/callbacks/inProgress"
+                ),
+                "failedUri": (
+                    "http://localhost:8000/integrations/ogc-api-processes/gep/callbacks/failed"
+                ),
+            },
         },
         _headers={
             "accept": "*/*",
@@ -530,6 +541,17 @@ async def test_execute_job_returns_plain_job_id_without_namespace(
                 "application": "buffer",
                 "user_id": "user-123",
             },
+            "subscriber": {
+                "successUri": (
+                    "http://localhost:8000/integrations/ogc-api-processes/gep/callbacks/success"
+                ),
+                "inProgressUri": (
+                    "http://localhost:8000/integrations/ogc-api-processes/gep/callbacks/inProgress"
+                ),
+                "failedUri": (
+                    "http://localhost:8000/integrations/ogc-api-processes/gep/callbacks/failed"
+                ),
+            },
         },
         _headers={
             "accept": "*/*",
@@ -576,6 +598,17 @@ async def test_execute_job_omits_missing_optional_user_fields(
         execute={
             "inputs": {"limit": 10},
             "properties": {"title": "My job", "application": "buffer"},
+            "subscriber": {
+                "successUri": (
+                    "http://localhost:8000/integrations/ogc-api-processes/gep/callbacks/success"
+                ),
+                "inProgressUri": (
+                    "http://localhost:8000/integrations/ogc-api-processes/gep/callbacks/inProgress"
+                ),
+                "failedUri": (
+                    "http://localhost:8000/integrations/ogc-api-processes/gep/callbacks/failed"
+                ),
+            },
         },
         _headers={
             "accept": "*/*",
diff --git a/tests/services/test_processing.py b/tests/services/test_processing.py
index 45f1ad7..641cbc6 100644
--- a/tests/services/test_processing.py
+++ b/tests/services/test_processing.py
@@ -66,7 +66,7 @@ async def test_create_processing_job_calls_platform_execute(
         id=fake_result,
         title=fake_job.title,
         label=ProcessTypeEnum.OPENEO,
-        status=ProcessingStatusEnum.CREATED,
+        status=ProcessingStatusEnum.QUEUED,
         parameters=fake_job.parameters,
         service=fake_job.service,
     )
@@ -74,7 +74,7 @@ async def test_create_processing_job_calls_platform_execute(
         id=fake_result,
         title=fake_summary.title,
         label=ProcessTypeEnum.OPENEO,
-        status=ProcessingStatusEnum.CREATED,
+        status=ProcessingStatusEnum.QUEUED,
     )
     fake_platform = MagicMock()
 
@@ -100,7 +100,7 @@ async def test_create_processing_job_calls_platform_execute(
     mock_save_job_to_db.assert_called_once()
     args, _ = mock_save_job_to_db.call_args
     saved_record = args[1]
-    assert saved_record.status == ProcessingStatusEnum.CREATED
+    assert saved_record.status == ProcessingStatusEnum.QUEUED
     assert result == fake_summary