[cross-repo from server#244] Conformance: run signals/queries against current published artifacts (#597)

Author: durable-workflow-ops

docs/api-stability.md

@@ -21,7 +21,7 @@ surface has *which* machine-readable spec, *what format* the spec uses,
 - <https://durable-workflow.github.io/docs/2.0/platform-protocol-specs>
 - machine-readable mirror: `platform_protocol_specs` in
   `GET /api/cluster/info`, schema
-  `durable-workflow.v2.platform-protocol-specs.catalog`, version `1`.
+  `durable-workflow.v2.platform-protocol-specs.catalog`, version `13`.
 - in-process source: `Workflow\V2\Support\PlatformProtocolSpecs`, which
   the standalone server re-exports verbatim.
 
@@ -179,6 +179,7 @@ Laravel queue runner:
 
 - `Workflow\V2\Worker\WorkerProtocolClient`
 - `Workflow\V2\Worker\WorkflowFiberRunner`
+- `Workflow\V2\Worker\WorkflowQueryTaskExecutor`
 - `Workflow\V2\Worker\WorkflowStep`
 
 These classes are covered by the same semver rules as the server-facing
@@ -197,13 +198,20 @@ history and has no terminal outcome yet, the step contains no commands; the
 worker must wait for a later history payload instead of duplicating the
 schedule command.
 `WorkerProtocolClient` defaults to the standalone server worker API:
-registration, heartbeat, workflow-task polling/history/complete/fail, and
-activity-task polling/heartbeat/complete/fail all use `POST /api/worker/...`
-with the worker-protocol headers. Standalone `poll*` methods return leased
-tasks from the server's `task` envelope; the client caches the returned
-lease fields so follow-up history, heartbeat, complete, and fail calls can
-send the required `lease_owner`, `workflow_task_attempt`, and
-`activity_attempt_id` values.
+registration, heartbeat, workflow-task polling/history/complete/fail,
+activity-task polling/heartbeat/complete/fail, and query-task
+polling/complete/fail all use `POST /api/worker/...` with the
+worker-protocol headers. Standalone `poll*` methods return leased tasks from
+the server's `task` envelope; the client caches the returned lease fields so
+follow-up history, heartbeat, complete, and fail calls can send the required
+`lease_owner`, `workflow_task_attempt`, `activity_attempt_id`, and
+`query_task_attempt` values.
+`WorkflowQueryTaskExecutor` is the stable PHP worker shim for the
+`query_tasks` capability. It accepts the server-routed query task envelope,
+replays the supplied history export in query mode, validates query targets
+and arguments against the same contract as `WorkflowStub::queryWithArguments`,
+and returns either an encoded result envelope or a typed query failure for
+`WorkerProtocolClient` to complete or fail the query task.
 Embedded package installs that need the `/webhooks` bridge contract must opt
 into embedded bridge mode explicitly; in that mode `poll*` methods return
 ready task opportunities as `tasks` lists and workers explicitly claim a task

docs/architecture/query-and-live-debug.md

@@ -164,6 +164,52 @@ the caller (for example `'run'` versus `'instance-current'`) so
 downstream clients can distinguish a query against a specific
 run from a query against the current run for an instance.
 
+## Worker-routed query tasks
+
+Standalone workers MAY execute queries through the worker-plane
+query task protocol instead of in the caller's HTTP process. This
+surface is part of the same non-durable query contract and is
+advertised by `Workflow\V2\Support\WorkerProtocolVersion::describe()`
+under `query_task_verbs` and `query_tasks`.
+
+Workers that can execute server-routed queries advertise the
+`query_tasks` worker capability when registering. The server may then
+lease query work through:
+
+- `POST /api/worker/query-tasks/poll`
+- `POST /api/worker/query-tasks/{query_task_id}/complete`
+- `POST /api/worker/query-tasks/{query_task_id}/fail`
+
+The poll request names `worker_id` and `task_queue`. A successful
+poll leases at most one query task and returns `poll_status =
+'leased'`; an empty long-poll returns `poll_status = 'empty'` with
+no task. Query task long-poll timeout semantics are the same
+clamped `WorkerProtocolVersion::longPollSemantics()` used by
+workflow and activity task polling.
+
+Each leased query task carries `query_task_id`,
+`query_task_attempt`, `lease_owner`, `workflow_id`, `run_id`,
+`query_name`, payload codec information, query arguments, and either
+a `history_export` snapshot or enough history fields for the worker
+to reconstruct one. The worker replays committed history with
+commands suppressed, invokes the declared query method, and then
+returns exactly one terminal query-task outcome.
+
+Completion requires `lease_owner`, `query_task_attempt`, `result`,
+and optionally `result_envelope` with `codec`, `blob`, or
+`external_storage`. Failure requires `lease_owner`,
+`query_task_attempt`, and `failure` with `message` plus optional
+`reason`, `type`, `stack_trace`, and `validation_errors`; known SDK
+reasons include `rejected_unknown_query`, `invalid_query_arguments`,
+and `query_rejected`.
+
+Completing or failing a query task resolves the waiting query
+request only. It MUST NOT append a workflow history event, create a
+workflow command, dispatch a wake notification for workflow work, or
+mutate the run. Adding, removing, or reshaping any query-task verb,
+field, or outcome is a worker protocol change and requires a
+`WorkerProtocolVersion::VERSION` bump.
+
 ## Non-durability guarantees
 
 A successful query MUST leave zero durable state behind.

src/V2/Support/PlatformProtocolSpecs.php

@@ -38,7 +38,7 @@ final class PlatformProtocolSpecs
 {
     public const SCHEMA = 'durable-workflow.v2.platform-protocol-specs.catalog';
 
-    public const VERSION = 12;
+    public const VERSION = 13;
 
     public const AUTHORITY_URL = 'https://durable-workflow.github.io/docs/2.0/platform-protocol-specs';
 
@@ -227,7 +227,7 @@ private static function specs(): array
                 'spec_path' => 'static/platform-protocol-specs/control-plane-api.openapi.yaml',
             ],
             'worker_protocol_api' => [
-                'description' => 'OpenAPI specification for the worker-plane HTTP+JSON API: register, poll, heartbeat, worker-session lifecycle, complete, and fail for workflow and activity tasks. The companion AsyncAPI document `worker_protocol_stream` describes the long-poll and lease-renewal semantics.',
+                'description' => 'OpenAPI specification for the worker-plane HTTP+JSON API: register, poll, heartbeat, worker-session lifecycle, complete, and fail for workflow, activity, and query tasks. Query-task routes are lease-fenced request/response work and are advertised through the `query_tasks` worker capability. The companion AsyncAPI document `worker_protocol_stream` describes the long-poll and lease-renewal semantics.',
                 'format' => self::FORMAT_OPENAPI,
                 'spec_id' => 'durable-workflow.v2.worker-protocol-api',
                 'surface_family' => 'worker_protocol',
@@ -253,6 +253,18 @@ private static function specs(): array
                         'schema_authority' => 'App\\Http\\Controllers\\Api\\WorkerController task completion/failure actions',
                         'version_authority' => 'Workflow\\V2\\Support\\WorkerProtocolVersion::VERSION',
                     ],
+                    [
+                        'name' => 'worker_query_task_poll_request',
+                        'owner_repo' => 'durable-workflow/server',
+                        'schema_authority' => 'App\\Http\\Controllers\\Api\\WorkerController::pollQueryTasks and App\\Support\\WorkflowQueryTaskBroker::poll',
+                        'version_authority' => 'Workflow\\V2\\Support\\WorkerProtocolVersion::VERSION',
+                    ],
+                    [
+                        'name' => 'worker_query_task_result',
+                        'owner_repo' => 'durable-workflow/server',
+                        'schema_authority' => 'App\\Http\\Controllers\\Api\\WorkerController::completeQueryTask/failQueryTask and App\\Support\\WorkflowQueryTaskBroker terminal outcomes',
+                        'version_authority' => 'Workflow\\V2\\Support\\WorkerProtocolVersion::VERSION',
+                    ],
                     [
                         'name' => 'external_task_input_contract',
                         'owner_repo' => 'durable-workflow/server',
@@ -269,7 +281,7 @@ private static function specs(): array
                 'evolution_rule' => self::EVOLUTION_ADDITIVE_MINOR_BREAKING_MAJOR,
                 'breaking_change_release' => 'major',
                 'discovery_endpoint' => 'GET /api/cluster/info -> worker_protocol',
-                'conformance_test' => 'durable-workflow/server: tests/Feature/WorkerProtocolContractTest.php, tests/Feature/WorkerProtocolSuccessContractTest.php, tests/Feature/WorkerProtocolOwnershipErrorContractTest.php, tests/Feature/WorkerProtocolVersionCoverageTest.php, tests/Feature/WorkflowWorkerProtocolTest.php, and tests/Feature/ActivityWorkerProtocolTest.php',
+                'conformance_test' => 'durable-workflow/server: tests/Feature/WorkerProtocolContractTest.php, tests/Feature/WorkerProtocolSuccessContractTest.php, tests/Feature/WorkerProtocolOwnershipErrorContractTest.php, tests/Feature/WorkerProtocolVersionCoverageTest.php, tests/Feature/WorkflowWorkerProtocolTest.php, tests/Feature/ActivityWorkerProtocolTest.php, and tests/Feature/WorkflowQueryTaskBrokerTest.php; durable-workflow/workflow: tests/Unit/V2/WorkerProtocolClientTest.php and tests/Unit/V2/WorkflowQueryTaskExecutorTest.php',
                 'status' => self::STATUS_PUBLISHED,
                 'spec_path' => 'static/platform-protocol-specs/worker-protocol-api.openapi.yaml',
             ],

src/V2/Support/SurfaceStabilityContract.php

@@ -177,7 +177,7 @@ private static function surfaceFamilies(): array
                 'notes' => 'Per-route version is governed by the `control_plane.request_contract` and `control_plane.response.contract` schema/version pairs published from `/api/cluster/info`. The top-level server `version` is build identity, not the client compatibility authority.',
             ],
             'worker_protocol' => [
-                'description' => 'Worker-plane HTTP API used by external SDK workers to register, poll, heartbeat, complete, and fail workflow and activity tasks.',
+                'description' => 'Worker-plane HTTP API used by external SDK workers to register, poll, heartbeat, complete, and fail workflow, activity, and query tasks.',
                 'stability_level' => self::STABILITY_STABLE,
                 'authority_manifest' => 'worker_protocol',
                 'requires_protocol_header' => 'X-Durable-Workflow-Protocol-Version',

src/V2/Support/WorkerProtocolVersion.php

@@ -29,7 +29,14 @@ final class WorkerProtocolVersion
      * pagination semantics). Bump the minor for additive changes (new
      * optional fields, new non-terminal command types).
      */
-    public const VERSION = '1.5';
+    public const VERSION = '1.6';
+
+    /**
+     * Worker registration capability for server-routed workflow query
+     * tasks. Workers that advertise this capability may receive query work
+     * through the query task poll/complete/fail endpoints.
+     */
+    public const CAPABILITY_QUERY_TASKS = 'query_tasks';
 
     /**
      * Stable fail-closed reason a worker or server must return when it
@@ -128,6 +135,31 @@ public static function activityTaskVerbs(): array
         return ['poll', 'claim', 'claimStatus', 'complete', 'fail', 'status', 'heartbeat'];
     }
 
+    /**
+     * Query task verbs exposed by the standalone worker protocol.
+     *
+     * Query tasks are server-routed, lease-fenced request/response work
+     * items. They replay committed history in a worker process, return a
+     * query result or typed failure to the waiting caller, and never append
+     * workflow history.
+     *
+     * @return list<string>
+     */
+    public static function queryTaskVerbs(): array
+    {
+        return ['poll', 'complete', 'fail'];
+    }
+
+    /**
+     * Worker registration capabilities with protocol-defined semantics.
+     *
+     * @return list<string>
+     */
+    public static function workerCapabilities(): array
+    {
+        return [self::CAPABILITY_QUERY_TASKS];
+    }
+
     /**
      * Non-terminal command types that an external worker may return
      * from a workflow task completion.
@@ -238,6 +270,8 @@ public static function taskQueuePriorityFairnessSemantics(): array
      *     version: string,
      *     workflow_task_verbs: list<string>,
      *     activity_task_verbs: list<string>,
+     *     query_task_verbs: list<string>,
+     *     worker_capabilities: list<string>,
      *     non_terminal_command_types: list<string>,
      *     terminal_command_types: list<string>,
      *     history_pagination: array{default_page_size: int, max_page_size: int},
@@ -247,6 +281,7 @@ public static function taskQueuePriorityFairnessSemantics(): array
      *     worker_session_verbs: list<string>,
      *     sticky_execution: array<string, mixed>,
      *     worker_sessions: array<string, mixed>,
+     *     query_tasks: array<string, mixed>,
      *     invocable_carrier: array<string, mixed>,
      *     task_queue_priority_fairness: array<string, mixed>,
      * }
@@ -257,6 +292,8 @@ public static function describe(): array
             'version' => self::VERSION,
             'workflow_task_verbs' => self::workflowTaskVerbs(),
             'activity_task_verbs' => self::activityTaskVerbs(),
+            'query_task_verbs' => self::queryTaskVerbs(),
+            'worker_capabilities' => self::workerCapabilities(),
             'non_terminal_command_types' => self::nonTerminalCommandTypes(),
             'terminal_command_types' => self::terminalCommandTypes(),
             'history_pagination' => [
@@ -272,6 +309,7 @@ public static function describe(): array
             'worker_session_verbs' => self::workerSessionVerbs(),
             'sticky_execution' => StickyExecution::describe(),
             'worker_sessions' => self::workerSessionSemantics(),
+            'query_tasks' => self::queryTaskSemantics(),
             'payload_codecs_universal' => CodecRegistry::universal(),
             'payload_codecs_engine_specific' => CodecRegistry::engineSpecific(),
             'unsupported_payload_codec_reason' => self::REASON_UNSUPPORTED_PAYLOAD_CODEC,
@@ -280,6 +318,86 @@ public static function describe(): array
         ];
     }
 
+    /**
+     * Published server-routed workflow query task contract.
+     *
+     * @return array<string, mixed>
+     */
+    public static function queryTaskSemantics(): array
+    {
+        return [
+            'feature' => self::CAPABILITY_QUERY_TASKS,
+            'minimum_protocol_version' => self::VERSION,
+            'worker_capability' => self::CAPABILITY_QUERY_TASKS,
+            'verbs' => self::queryTaskVerbs(),
+            'path_prefix' => '/api/worker/query-tasks',
+            'endpoints' => [
+                'poll' => [
+                    'method' => 'POST',
+                    'path' => '/api/worker/query-tasks/poll',
+                    'request_fields' => ['worker_id', 'task_queue'],
+                    'response_fields' => ['task', 'poll_status'],
+                ],
+                'complete' => [
+                    'method' => 'POST',
+                    'path' => '/api/worker/query-tasks/{query_task_id}/complete',
+                    'request_fields' => [
+                        'lease_owner',
+                        'query_task_attempt',
+                        'result',
+                        'result_envelope',
+                    ],
+                ],
+                'fail' => [
+                    'method' => 'POST',
+                    'path' => '/api/worker/query-tasks/{query_task_id}/fail',
+                    'request_fields' => ['lease_owner', 'query_task_attempt', 'failure'],
+                ],
+            ],
+            'poll' => [
+                'leases_on_return' => true,
+                'long_poll' => self::longPollSemantics(),
+                'empty_response_poll_status' => 'empty',
+                'requires_registered_worker' => true,
+                'requires_worker_capability' => self::CAPABILITY_QUERY_TASKS,
+            ],
+            'task_fields' => [
+                'query_task_id',
+                'query_task_attempt',
+                'lease_owner',
+                'workflow_id',
+                'run_id',
+                'task_queue',
+                'workflow_type',
+                'workflow_class',
+                'query_name',
+                'query_arguments',
+                'payload_codec',
+                'history_export',
+                'history_events',
+            ],
+            'completion' => [
+                'requires_lease_owner' => true,
+                'requires_query_task_attempt' => true,
+                'result_envelope_fields' => ['codec', 'blob', 'external_storage'],
+                'terminal_for_query_task' => true,
+            ],
+            'failure' => [
+                'requires_lease_owner' => true,
+                'requires_query_task_attempt' => true,
+                'failure_fields' => ['message', 'reason', 'type', 'stack_trace', 'validation_errors'],
+                'known_reasons' => ['rejected_unknown_query', 'invalid_query_arguments', 'query_rejected'],
+                'terminal_for_query_task' => true,
+            ],
+            'durability' => [
+                'history_event_appended' => false,
+                'workflow_command_created' => false,
+                'result_resolves_waiting_query_request' => true,
+                'query_replay_must_suppress_commands' => true,
+            ],
+        ];
+    }
+
     /**
      * Published invocable HTTP carrier wire-protocol contract.
      *