feat: add event-driven job status callbacks for OGC API Processes execution

[sim13pods]

APEx and the target OGC API Processes platform maintain separate records of a job’s status, once APEx submits a job, the target platform executes it independently and updates its own status as processing progresses and these changes are not automatically reflected in the status stored by APEx.

Currently, rebuilding the full API call stack, as shown in the sequence diagram below, APEx refreshes its stored job status only when a client invokes GET /jobs_status. This means that synchronization depends on a client requesting an update, rather than on the job actually changing state. Between requests—or if no further requests are made—the status recorded by APEx can become outdated.
For example, a job may have already completed, failed, or been canceled on the target platform while APEx still records it as queued or running. This discrepancy does not necessarily indicate an execution problem: it means that APEx has not yet learned about the change. As a result, users and services relying on APEx’s stored status may receive an inaccurate picture of ongoing processing, and APEx cannot provide timely notifications of job status changes.

As a result:

• Job statuses may remain out of sync (e.g., a failed job can remain stored as RUNNING until a client queries its status);
• Under these conditions, the APEx Dispatcher API cannot emit near real-time status change events.

This PR addresses the gap by allowing the target platform to notify APEx when a job progresses or reaches a final state. APEx updates its stored status when these notifications arrive, reducing the delay between execution changes and their visibility in APEx, without requiring a client to request a status refresh.

Solution description

Without drastically redesigning the APEx Dispatcher API, our solution builds upon the existing OGC API - Processes execute API by extending its execution request payload with the subscriber element:

POST /processes/{processID}/execution
{
  ...,
  "subscriber": {
    "successUri": "string",
    "inProgressUri": "string",
    "failedUri": "string"
  }
}

This PR enables the Dispatch API to receive job lifecycle notifications from execution runners and update the stored APEx job status. OGC execution requests now include subscriber callback URLs, allowing runners to report progress, completion, and failure directly to the dispatcher.

The callback payloads use a common envelope based on the CloudEvents core specification and its Partitioning extension. Building on this envelope, the Stage Events event models define the event types and stage-specific data, giving runners and the dispatcher a shared, explicitly defined contract for each execution stage. Reusing their schema-generated Pydantic models provides validation of required fields and event types at the API boundary, reducing duplicated schema maintenance and the risk of producers and consumers interpreting the same notification differently. The common CloudEvent envelope carries the event type, source, and workflow subject, providing consistent context for identifying the originating execution and correlating its notifications. Stage-specific payloads retain execution detail, such as inputs, outputs, usage, and optional failure information, while the dispatcher maps supported events to its existing APEx status model. This keeps runner event semantics separate from the dispatcher's status representation and provides a foundation for future monitoring consumers to reuse the same payload contract. Together, the OGC subscriber mechanism and the Stage Events payloads address complementary concerns: where notifications are delivered and how their content is interpreted. Delivered callbacks can therefore update stored job status without waiting for a client status query, using an existing event model without requiring a database schema change.

The implementation adds three endpoints accepting structured CloudEvents:

Endpoint	Supported events	APEx status
POST /unit_jobs/inProgress	submitted	QUEUED
POST /unit_jobs/inProgress	prepared, staged	RUNNING
POST /unit_jobs/inProgress	dismissed	CANCELED
POST /unit_jobs/success	piped, completed	FINISHED
POST /unit_jobs/failed	failed	FAILED

Callbacks identify the job through the CloudEvent subject, without requiring a job ID in the callback URL. Three-part subjects (<namespace>:<workflow_uid>:<workflow_name>) are normalized to <namespace>:<workflow_name>. The lookup service supports internal numeric IDs and platform job identifiers, including workflow-name matching.

Successful callbacks return 200 OK. Progress notifications received before the job is registered are also acknowledged with 200 OK, without updating a record, to accommodate notifications arriving during job creation. Success and failure callbacks for unknown jobs return 404, while unexpected errors during callback lookup or status updates return 500.

These callback errors are returned as application/problem+json, the JSON representation defined by RFC 9457 — Problem Details for HTTP APIs, which supersedes RFC 7807. The standard complements HTTP status codes with a machine-readable problem description: type identifies the problem category, title provides a short summary, status records the HTTP status, detail provides explanatory text, and instance identifies the problem occurrence. This gives callback senders a predictable response structure and allows them to classify errors by problem type without parsing human-readable messages.

The implementation uses the EOAP Problems Registry to supply shared problem definitions and their documented type URIs. Its NotFound and ServerError models, serialized through the registry's FastAPI exception handler, keep callback responses aligned with the same error vocabulary used by other APIs adopting the registry. The dispatcher sets instance to the callback request path and adds job-specific information through the registry's errors extension, with parameter="job_id" and a descriptive detail. Here, job_id labels the affected job identifier extracted from the CloudEvent subject; it is not a callback URL parameter. Reusing these models centralizes the response definitions, reduces duplicated error-handling code, and gives clients and monitoring tools a consistent basis for reporting and diagnosing callback failures.

Solution implementation details

Additional changes include:

• Adding successUri, inProgressUri, and failedUri to the subscriber payload sent with OGC execution requests.
• Introducing DISPATCH_API_PUBLIC_URL to construct callback URLs.
• Changing the initial status of successfully submitted jobs from CREATED to QUEUED, and mapping OGC accepted status to QUEUED.
• Documenting callback endpoints, CloudEvent payloads, and problem responses in the OpenAPI specification.

Tests added or updated cover event-to-status mappings, subject normalization, job lookup and status updates, unknown-job problem responses, early progress acknowledgements, and subscriber payload generation.

For deployment, set DISPATCH_API_PUBLIC_URL to the externally reachable Dispatch API base URL, including any deployment path prefix. It defaults to http://localhost:8000.

No database schema migration is required.

Callback invocation examples

Set the Dispatch API base URL, including any deployment path prefix:

DISPATCH_API_BASE_URL="http://localhost:8000"

The examples below use the workflow subject default:test-uid:job-123, which the dispatcher resolves to the platform job identifier default:job-123. Replace these values with an existing test job's namespace, workflow UID, and workflow name. Keep partitionkey aligned with subject and update the identifiers in data accordingly. Each example includes a distinct sample event id; use a new ID for each distinct event you produce.

These are individual examples, not a script to run sequentially: success, failure, and dismissal represent different outcomes. Each invocation updates the matching job's stored status. Empty objects illustrate the payload shapes used in the callback tests; real runners can supply the corresponding inputs, outputs, entries, and usage information.

submitted → QUEUED

curl --include --request POST \
  "${DISPATCH_API_BASE_URL%/}/unit_jobs/inProgress" \
  --header "Content-Type: application/cloudevents+json" \
  --data-binary @- <<'JSON'
{
  "specversion": "1.0",
  "id": "example-job-123-submitted-001",
  "source": "terradue:apex:test",
  "subject": "default:test-uid:job-123",
  "partitionkey": "default:test-uid:job-123",
  "type": "submitted",
  "datacontenttype": "application/json",
  "data": {
    "namespace": "default",
    "process_id": "test-proc"
  }
}
JSON

prepared → RUNNING

curl --include --request POST \
  "${DISPATCH_API_BASE_URL%/}/unit_jobs/inProgress" \
  --header "Content-Type: application/cloudevents+json" \
  --data-binary @- <<'JSON'
{
  "specversion": "1.0",
  "id": "example-job-123-prepared-001",
  "source": "terradue:apex:test",
  "subject": "default:test-uid:job-123",
  "partitionkey": "default:test-uid:job-123",
  "type": "prepared",
  "datacontenttype": "application/json",
  "data": {
    "namespace": "default",
    "process_id": "test-proc",
    "job_id": "job-123",
    "inputs": {}
  }
}
JSON

staged → RUNNING

curl --include --request POST \
  "${DISPATCH_API_BASE_URL%/}/unit_jobs/inProgress" \
  --header "Content-Type: application/cloudevents+json" \
  --data-binary @- <<'JSON'
{
  "specversion": "1.0",
  "id": "example-job-123-staged-001",
  "source": "terradue:apex:test",
  "subject": "default:test-uid:job-123",
  "partitionkey": "default:test-uid:job-123",
  "type": "staged",
  "datacontenttype": "application/json",
  "data": {
    "namespace": "default",
    "process_id": "test-proc",
    "job_id": "job-123",
    "outputs": {}
  }
}
JSON

piped → FINISHED

curl --include --request POST \
  "${DISPATCH_API_BASE_URL%/}/unit_jobs/success" \
  --header "Content-Type: application/cloudevents+json" \
  --data-binary @- <<'JSON'
{
  "specversion": "1.0",
  "id": "example-job-123-piped-001",
  "source": "terradue:apex:test",
  "subject": "default:test-uid:job-123",
  "partitionkey": "default:test-uid:job-123",
  "type": "piped",
  "datacontenttype": "application/json",
  "data": {
    "namespace": "default",
    "process_id": "test-proc",
    "job_id": "job-123",
    "entries": {}
  }
}
JSON

completed → FINISHED

curl --include --request POST \
  "${DISPATCH_API_BASE_URL%/}/unit_jobs/success" \
  --header "Content-Type: application/cloudevents+json" \
  --data-binary @- <<'JSON'
{
  "specversion": "1.0",
  "id": "example-job-123-completed-001",
  "source": "terradue:apex:test",
  "subject": "default:test-uid:job-123",
  "partitionkey": "default:test-uid:job-123",
  "type": "completed",
  "datacontenttype": "application/json",
  "data": {
    "namespace": "default",
    "process_id": "test-proc",
    "job_id": "job-123",
    "outputs": {},
    "usage": {}
  }
}
JSON

failed → FAILED

curl --include --request POST \
  "${DISPATCH_API_BASE_URL%/}/unit_jobs/failed" \
  --header "Content-Type: application/cloudevents+json" \
  --data-binary @- <<'JSON'
{
  "specversion": "1.0",
  "id": "example-job-123-failed-001",
  "source": "terradue:apex:test",
  "subject": "default:test-uid:job-123",
  "partitionkey": "default:test-uid:job-123",
  "type": "failed",
  "datacontenttype": "application/json",
  "data": {
    "namespace": "default",
    "process_id": "test-proc",
    "job_id": "job-123",
    "usage": {}
  }
}
JSON

dismissed → CANCELED

curl --include --request POST \
  "${DISPATCH_API_BASE_URL%/}/unit_jobs/inProgress" \
  --header "Content-Type: application/cloudevents+json" \
  --data-binary @- <<'JSON'
{
  "specversion": "1.0",
  "id": "example-job-123-dismissed-001",
  "source": "terradue:apex:test",
  "subject": "default:test-uid:job-123",
  "partitionkey": "default:test-uid:job-123",
  "type": "dismissed",
  "datacontenttype": "application/json",
  "data": {
    "namespace": "default",
    "status": "Pending"
  }
}
JSON

For dismissed, the payload's data.status is the runner status (Pending in this example); the event type determines the APEx mapping to CANCELED.

For an existing job, each successful invocation returns 200 OK. An inProgress callback for an unknown job also returns 200 OK, but no record is updated. Unknown jobs on the success and failed endpoints return 404 with an application/problem+json response; unexpected lookup or update errors return 500 in the same format.

The sequence diagram below illustrates the complete flow, from registering callback URLs during job submission to updating the stored APEx job status when notifications arrive:

Co-authored-by: maxmiraftab max.miraftab@terradue.com

Appendix: Reusing job status notifications across target platforms

Integrators of other APEx target platforms are encouraged to adopt the same notification endpoints and payload format introduced by this PR. Although the integration described here uses OGC API Processes to register callback URLs, the notification mechanism provides a common way for target platforms to report job progress and outcomes to APEx.

By sending notifications when a job changes state, a platform can help APEx maintain an up-to-date view of its execution without waiting for a client to request a status refresh. Extending this approach across integrations would provide users with more consistent status reporting, regardless of which platform executes their jobs.

Integrations should reuse the existing endpoints:

• POST /unit_jobs/inProgress for submission, progress, and cancellation notifications.
• POST /unit_jobs/success for successful execution.
• POST /unit_jobs/failed for failed execution.

Notifications should follow the same CloudEvents envelope and supported Stage Events payload schemas, including the job reference used by APEx to associate each notification with the correct execution. Platforms with different internal lifecycle models should map their states to the supported events, preserving the shared meaning of each event and its corresponding APEx status.

Reusing this contract avoids introducing a separate notification format or additional endpoints for every platform. It also allows validation, status mapping, and error handling to remain consistent across integrations, while providing a shared foundation for future monitoring capabilities.

Integrators are therefore invited to use the payload definitions and invocation examples in this PR as the starting point for their implementation. Where a platform’s lifecycle cannot be represented by the existing events, we encourage discussing extensions to the shared contract so that improvements can benefit all participating platforms.