Add typed workflow version-usage reads

CHANGELOG.md

@@ -6,6 +6,13 @@ project follows [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Added
+
+- `Client::workflowVersionUsage()` reads typed live v1 open-run version usage,
+  with all-state counts, explicit unknown reasons and opaque continuation cursors
+  (.github#220). It requires the advertised `workflow_version_usage` capability
+  and retains the control credential, namespace and request timeout policy.
+
 ### Security
 
 - Raise the optional Laravel bridge floors to 12.72.0 and 13.41.0 for

docs/portal/build/workflows-activities.md

@@ -407,6 +407,49 @@ A later deployment can raise the maximum while keeping the recorded versions in
 
 For a boolean rollout, use `patched('require-reservation-review')`. It returns `false` for legacy histories and `true` for new histories. After the legacy branch is no longer supported, replace the conditional call with `deprecatePatch()` for the same ID before eventually removing the marker. Compatible numeric and patch helpers share the same frozen version for a change ID. Every call must support that version, and `patched()` returns whether it equals `1`. Do not reuse a change ID for an unrelated rollout.
 
+### Inspect open-run version usage
+
+Before raising a minimum or removing a legacy branch, inspect the open runs for
+that exact workflow type and change ID. `workflowVersionUsage()` requires the
+Server's `workflow_version_usage` capability and throws
+`ServerCapabilityUnavailable` when it is absent. Discovery and the metadata read
+use the client's control credential, namespace and configured control-request
+timeout. They do not use a worker credential or fetch application payloads.
+
+```php
+$page = $client->workflowVersionUsage(
+    workflowType: 'shipment',
+    changeId: 'carrier-api',
+    minVersion: 2,
+    state: 'legacy',
+    limit: 50,
+);
+
+// Counts include every state, even though this page lists only legacy runs.
+$unknown = $page->counts->unknown;
+foreach ($page->runs as $run) {
+    printf("%s %s %s\n", $run->instanceId, $run->runId, $run->state);
+}
+
+// Repeat the exact namespace/type/change/minimum/state query to continue.
+$nextCursor = $page->nextCursor;
+```
+
+`minVersion` is the proposed minimum supported signed 32-bit version, including
+negative legacy sentinels; it is not the largest deployed version. The response
+distinguishes `legacy`, `current` and `unknown`. Missing markers, unavailable or
+invalid history remain unknown with their original reason. A current build or
+definition is not evidence that a run recorded the current branch.
+
+Each page is a live observation over a fixed upper run-ID bound, not a durable
+fleet snapshot. Runs can close or record decisions between pages, so counts can
+change. Pass cursors unchanged with the exact same filters; invalid cursors are
+errors, not a reason to restart silently. Unknowns remain relevant to rollout
+decisions. Fence or drain admissions and restart the traversal before making a
+removal decision; this API returns no `safe_to_remove` authorization. The
+[shared v1 contract](https://github.com/durable-workflow/workflow/blob/c93cadad8a0104b6e7e5f38e022c3b06c26c86e6/docs/version-usage.md)
+defines the classification and live paging limits.
+
 A `waitCondition()` predicate is part of replay identity too. New waits use a
 `dwphp2:sha256:` fingerprint of the callable's tokens. Whitespace and comments
 outside literals, file locations and neighboring call-site arguments are

src/Client.php

@@ -39,6 +39,7 @@
 use DurableWorkflow\Model\ServiceOperationOptions;
 use DurableWorkflow\Model\WorkflowExecution;
 use DurableWorkflow\Model\WorkflowPage;
+use DurableWorkflow\Model\WorkflowVersionUsagePage;
 use DurableWorkflow\Model\WorkflowRun;
 use DurableWorkflow\Model\WorkflowStreamAppendItem;
 use DurableWorkflow\Model\WorkflowStreamAppendResult;
@@ -388,6 +389,57 @@ public function clusterInfo(bool $includeDiagnostics = false): ClusterInfo
         return ClusterInfo::fromArray($info);
     }
 
+    /**
+     * Read one page of open-run version decisions for a proposed minimum supported version.
+     *
+     * Uses the existing control credential, namespace and control-request timeout policy.
+     * Cursors are opaque and query-bound: pass the returned nextCursor unchanged with the
+     * same filters. Counts precede state filtering and each page is a live observation,
+     * not a durable snapshot or authorization to remove a version branch.
+     *
+     * @throws ServerCapabilityUnavailable when workflow_version_usage is not advertised
+     * @throws \UnexpectedValueException when the v1 response or query identity is invalid
+     */
+    public function workflowVersionUsage(
+        string $workflowType,
+        string $changeId,
+        int $minVersion,
+        ?string $state = null,
+        int $limit = 50,
+        ?string $cursor = null,
+    ): WorkflowVersionUsagePage {
+        foreach ([[$workflowType, 255], [$changeId, 1024]] as [$identity, $maximum]) {
+            if (strlen($identity) > $maximum || preg_match('/[^\s\p{Z}]/u', $identity) !== 1) {
+                throw new InvalidArgumentException('Version usage needs nonblank UTF-8 workflow type and change ID within their byte limits.');
+            }
+        }
+        if ($minVersion < -2147483648 || $minVersion > 2147483647
+            || !in_array($state, [null, 'legacy', 'current', 'unknown'], true) || $limit < 1 || $limit > 200
+            || ($cursor !== null && ($cursor === '' || preg_match('//u', $cursor) !== 1))) {
+            throw new InvalidArgumentException('Invalid version-usage minimum, state, limit or opaque cursor.');
+        }
+        $info = ClusterInfo::fromArray($this->request('GET', self::CLUSTER_INFO_PATH, false,
+            payloadPaths: static fn (array $metadata): array => []));
+        if (!$info->supportsWorkflowVersionUsage()) {
+            throw new ServerCapabilityUnavailable(ClusterInfo::WORKFLOW_VERSION_USAGE,
+                'This Server does not advertise workflow_version_usage.');
+        }
+        $query = $this->withoutNulls([
+            'workflow_type' => $workflowType, 'change_id' => $changeId, 'min_version' => $minVersion,
+            'state' => $state, 'limit' => $limit, 'cursor' => $cursor,
+        ]);
+        $response = $this->request('GET', '/workflow-version-usage?'.http_build_query($query, '', '&', PHP_QUERY_RFC3986),
+            false, payloadPaths: static fn (array $metadata): array => []);
+        $page = WorkflowVersionUsagePage::fromArray($response, $cursor);
+        if ($page->namespace !== $this->namespace || $page->workflowType !== $workflowType
+            || $page->changeId !== $changeId || $page->minVersion !== $minVersion
+            || $page->state !== $state || $page->limit !== $limit) {
+            throw new \UnexpectedValueException('Version-usage response does not match the authenticated namespace and exact query.');
+        }
+
+        return $page;
+    }
+
     /**
      * @param array<string, int|string> $filters
      * @return array<string, mixed>

src/Metrics/MetricsRoutes.php

@@ -136,7 +136,7 @@ private static function sharedTaskRoutes(): array
     /** @return list<string> Reviewed operations only the PHP client currently sends. */
     private static function phpRoutes(): array
     {
-        $routes = ['/activities/{activity_id}/result', '/system/health', '/system/operator-dashboard',
+        $routes = ['/workflow-version-usage', '/activities/{activity_id}/result', '/system/health', '/system/operator-dashboard',
             '/system/operator-dashboard/bounded', '/system/operator-dashboard/bounded/workflow-types',
             '/system/operator-metrics', '/task-queues/{task_queue}/activity-dispatch',
             '/task-queues/{task_queue}/activity-dispatch/pause', '/task-queues/{task_queue}/activity-dispatch/resume'];

src/Model/ClusterInfo.php

@@ -13,6 +13,9 @@ final class ClusterInfo
     /** Signal and update `request_id` is a durable command key (native Server 3.9.0 and later). */
     public const COMMAND_REQUEST_KEYS = 'command_request_keys';
 
+    /** Read-only v1 open-run version usage; availability is discovered, never inferred from a version. */
+    public const WORKFLOW_VERSION_USAGE = 'workflow_version_usage';
+
     /**
      * @param array<string, mixed> $namespace
      * @param array<string, mixed> $capabilities
@@ -69,6 +72,11 @@ public function supportsCommandRequestKeys(): bool
         return ($this->capabilities[self::COMMAND_REQUEST_KEYS] ?? null) === true;
     }
 
+    public function supportsWorkflowVersionUsage(): bool
+    {
+        return ($this->capabilities[self::WORKFLOW_VERSION_USAGE] ?? null) === true;
+    }
+
     /** @return array<string, mixed> */
     private static function map(mixed $value): array
     {

src/Model/WorkflowVersionUsageCounts.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Model;
+
+use UnexpectedValueException;
+
+/** All-state counts for the live, upper-bounded open-run cohort before filtering. */
+final class WorkflowVersionUsageCounts
+{
+    private function __construct(
+        public readonly int $total,
+        public readonly int $legacy,
+        public readonly int $current,
+        public readonly int $unknown,
+    ) {
+    }
+
+    /** @param array<string, mixed> $value */
+    public static function fromArray(array $value): self
+    {
+        $keys = array_keys($value);
+        sort($keys);
+        if ($keys !== ['current', 'legacy', 'total', 'unknown']) {
+            throw new UnexpectedValueException('Invalid workflow version-usage counts.');
+        }
+        foreach ($value as $count) {
+            if (!is_int($count) || $count < 0) {
+                throw new UnexpectedValueException('Workflow version-usage counts must be non-negative integers.');
+            }
+        }
+        if ($value['total'] !== $value['legacy'] + $value['current'] + $value['unknown']) {
+            throw new UnexpectedValueException('Workflow version-usage counts do not reconcile.');
+        }
+
+        return new self($value['total'], $value['legacy'], $value['current'], $value['unknown']);
+    }
+}

src/Model/WorkflowVersionUsagePage.php

@@ -0,0 +1,137 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Model;
+
+use UnexpectedValueException;
+
+/** One live v1 observation; never a durable snapshot or permission to remove branches. */
+final class WorkflowVersionUsagePage
+{
+    public const SCHEMA_VERSION = 1;
+    public const CONSISTENCY = 'live';
+
+    /**
+     * @param list<WorkflowVersionUsageRun> $runs
+     * @param array<string, mixed> $raw
+     */
+    private function __construct(
+        public readonly ?string $namespace,
+        public readonly string $workflowType,
+        public readonly string $changeId,
+        public readonly int $minVersion,
+        public readonly ?string $state,
+        public readonly ?string $throughRunId,
+        public readonly WorkflowVersionUsageCounts $counts,
+        public readonly array $runs,
+        public readonly int $limit,
+        public readonly int $returnedCount,
+        public readonly bool $hasMore,
+        public readonly ?string $nextCursor,
+        public readonly array $raw,
+    ) {
+    }
+
+    /**
+     * @param array<string, mixed> $value
+     * @param string|null $requestCursor The opaque cursor supplied for this exact page request.
+     */
+    public static function fromArray(array $value, ?string $requestCursor = null): self
+    {
+        self::requireShape($value);
+        self::requireTextFields($value);
+        $counts = WorkflowVersionUsageCounts::fromArray($value['counts']);
+        if ($value['returned_count'] !== count($value['runs']) || $value['returned_count'] > $value['limit']
+            || $value['returned_count'] > $counts->total
+            || ($value['has_more'] !== ($value['next_cursor'] !== null))
+            || ($value['has_more'] && $value['returned_count'] === 0)
+            || ($value['through_run_id'] === null && $counts->total !== 0)) {
+            throw new UnexpectedValueException('Workflow version-usage page boundaries do not reconcile.');
+        }
+        $remaining = ['legacy' => $counts->legacy, 'current' => $counts->current, 'unknown' => $counts->unknown];
+        $selectedCount = $value['state'] === null ? $counts->total : $remaining[$value['state']];
+        if ($value['has_more'] && $selectedCount <= $value['returned_count']) {
+            throw new UnexpectedValueException('Workflow version-usage continuation has no remaining matching runs.');
+        }
+        self::requireCursorProgress($value, $selectedCount, $requestCursor);
+
+        return new self($value['namespace'], $value['workflow_type'], $value['change_id'], $value['min_version'],
+            $value['state'], $value['through_run_id'], $counts, self::readRuns($value, $remaining), $value['limit'], $value['returned_count'],
+            $value['has_more'], $value['next_cursor'], $value);
+    }
+
+    /** @param array<string, mixed> $value */
+    private static function requireCursorProgress(array $value, int $selectedCount, ?string $requestCursor): void
+    {
+        if ($requestCursor === null && !$value['has_more'] && $value['returned_count'] !== $selectedCount) {
+            throw new UnexpectedValueException('The first terminal version-usage page must contain the complete selected cohort.');
+        }
+        if ($value['has_more'] && $value['returned_count'] !== $value['limit']) {
+            throw new UnexpectedValueException('A version-usage continuation page must fill its requested limit.');
+        }
+        if ($requestCursor !== null && $value['next_cursor'] === $requestCursor) {
+            throw new UnexpectedValueException('A version-usage continuation cursor must advance.');
+        }
+    }
+
+    /** @param array<string, mixed> $value */
+    private static function requireShape(array $value): void
+    {
+        $keys = array_keys($value);
+        sort($keys);
+        if ($keys !== ['change_id', 'consistency', 'counts', 'has_more', 'limit', 'min_version', 'namespace',
+            'next_cursor', 'returned_count', 'runs', 'schema_version', 'state', 'through_run_id', 'workflow_type']
+            || $value['schema_version'] !== 1 || $value['consistency'] !== 'live'
+            || !is_int($value['min_version']) || $value['min_version'] < -2147483648 || $value['min_version'] > 2147483647
+            || !in_array($value['state'], [null, 'legacy', 'current', 'unknown'], true)
+            || !is_int($value['limit']) || $value['limit'] < 1 || $value['limit'] > 200
+            || !is_int($value['returned_count']) || $value['returned_count'] < 0
+            || !is_bool($value['has_more']) || !is_array($value['counts'])
+            || !is_array($value['runs']) || !array_is_list($value['runs'])) {
+            throw new UnexpectedValueException('Invalid workflow version-usage v1 response.');
+        }
+    }
+
+    /** @param array<string, mixed> $value */
+    private static function requireTextFields(array $value): void
+    {
+        foreach (['namespace', 'workflow_type', 'change_id', 'through_run_id', 'next_cursor'] as $field) {
+            if ($value[$field] === null && in_array($field, ['namespace', 'through_run_id', 'next_cursor'], true)) {
+                continue;
+            }
+            if (!is_string($value[$field]) || $value[$field] === '' || preg_match('//u', $value[$field]) !== 1) {
+                throw new UnexpectedValueException('Invalid workflow version-usage response identity or cursor.');
+            }
+        }
+    }
+
+    /**
+     * @param array<string, mixed> $value
+     * @param array<string, int> $remaining
+     * @return list<WorkflowVersionUsageRun>
+     */
+    private static function readRuns(array $value, array $remaining): array
+    {
+        $runs = [];
+        $previous = null;
+        foreach ($value['runs'] as $row) {
+            if (!is_array($row)) {
+                throw new UnexpectedValueException('Invalid workflow version-usage run.');
+            }
+            $run = WorkflowVersionUsageRun::fromArray($row, $value['min_version']);
+            if (--$remaining[$run->state] < 0) {
+                throw new UnexpectedValueException('Workflow version-usage page exceeds its all-state counts.');
+            }
+            if (($previous !== null && strcmp($previous, $run->runId) >= 0)
+                || $value['through_run_id'] === null || strcmp($run->runId, $value['through_run_id']) > 0
+                || ($value['state'] !== null && $run->state !== $value['state'])) {
+                throw new UnexpectedValueException('Workflow version-usage runs violate their order, upper bound or filter.');
+            }
+            $runs[] = $run;
+            $previous = $run->runId;
+        }
+
+        return $runs;
+    }
+}

src/Model/WorkflowVersionUsageRun.php

@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Model;
+
+use UnexpectedValueException;
+
+/** A recorded version decision, or an explicit unknown, for one open run. */
+final class WorkflowVersionUsageRun
+{
+    private function __construct(
+        public readonly string $instanceId,
+        public readonly string $runId,
+        public readonly string $status,
+        public readonly string $state,
+        public readonly ?int $recordedVersion,
+        public readonly ?string $unknownReason,
+    ) {
+    }
+
+    /** @param array<string, mixed> $value */
+    public static function fromArray(array $value, int $minimum): self
+    {
+        $keys = array_keys($value);
+        sort($keys);
+        if ($keys !== ['instance_id', 'recorded_version', 'run_id', 'state', 'status', 'unknown_reason']) {
+            throw new UnexpectedValueException('Invalid workflow version-usage run fields.');
+        }
+        foreach (['instance_id', 'run_id'] as $field) {
+            if (!is_string($value[$field]) || $value[$field] === '' || preg_match('//u', $value[$field]) !== 1) {
+                throw new UnexpectedValueException('Invalid workflow version-usage run identity.');
+            }
+        }
+        if (!in_array($value['status'], ['pending', 'running', 'waiting'], true)
+            || !in_array($value['state'], ['legacy', 'current', 'unknown'], true)) {
+            throw new UnexpectedValueException('Invalid workflow version-usage run state.');
+        }
+        $version = $value['recorded_version'];
+        $reason = $value['unknown_reason'];
+        if ($value['state'] === 'unknown') {
+            if ($version !== null || !in_array($reason, ['marker_absent', 'history_unavailable', 'history_invalid'], true)) {
+                throw new UnexpectedValueException('Unknown version usage requires its explicit unknown reason.');
+            }
+        } elseif (!is_int($version) || $version < -2147483648 || $version > 2147483647 || $reason !== null
+            || (($version < $minimum) !== ($value['state'] === 'legacy'))) {
+            throw new UnexpectedValueException('Recorded version usage does not match the proposed minimum.');
+        }
+
+        return new self($value['instance_id'], $value['run_id'], $value['status'], $value['state'], $version, $reason);
+    }
+}