Add namespace-scoped open-run version usage queries

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+- Add a namespace-scoped open-run version-usage query with legacy/current/unknown
+  counts and byte-exact keyset pages. Missing markers and unavailable history
+  remain unknown when assessing legacy branch retirement.
+
 - Experimental embedded compensation scopes (#1063): retained versioned
   obligations, native Saga recovery, sequential child participation and explicit
   release. Each scope keeps its originating run; continue-as-new, origin redrive

docs/version-usage.md

@@ -0,0 +1,125 @@
+# Open-run version usage (draft v1)
+
+This read-only contract answers which open runs still have a recorded decision
+below a proposed minimum supported
+version. It does not replay workflows, run application code, write a projection,
+change markers, or authorize deleting a branch.
+
+The native control-plane route is `GET /workflow-version-usage`, relative to the
+existing control-plane API root. Authentication and namespace selection use the
+existing runtime request context. The embedded entry point is
+`Workflow\V2\Support\VersionUsageQuery::page()`; its namespace argument is required,
+and `null` selects only the embedded default namespace, never all namespaces.
+Namespace, workflow type and change ID comparisons are exact and case-sensitive.
+
+## Request
+
+- `workflow_type`: required nonblank UTF-8 string, at most 255 bytes.
+- `change_id`: required nonblank UTF-8 string, at most 1024 bytes.
+- `min_version`: required signed 32-bit integer, from -2147483648 to 2147483647.
+  This is the proposed new `minSupported`, not the largest version deployed.
+  For removing a `patched()` legacy branch, use 1.
+- `state`: optional `legacy`, `current`, or `unknown`; omitted means all states.
+- `limit`: integer from 1 to 200, default 50.
+- `cursor`: optional opaque continuation from this exact namespace/type/change/
+  minimum/state query. It is not an authorization token. Invalid or mismatched
+  cursors are errors; they never fall back to a first page. Cursor bodies are
+  authenticated before parsing. The embedded reader uses a domain-separated HMAC
+  with the configured application key; missing keys are errors and key rotation
+  invalidates prior cursors.
+
+Invalid arguments return the existing invalid-request error envelope (HTTP 422
+for the native endpoint). Storage or connection failures are errors, not empty
+successful answers. Implementations must use the configured storage models and
+connection, including custom table names and prefixes.
+
+## Classification
+
+Only runs whose canonical status is open participate (embedded: pending,
+running, waiting). Retained closed, archived terminal and prior continued runs
+are excluded. An archived run which is still open remains included.
+
+- `legacy`: a retained, valid marker for this exact change ID has version less
+  than `min_version`.
+- `current`: its recorded version is greater than or equal to `min_version`.
+- `unknown`: no marker exists, or retained history cannot establish the decision.
+
+Each run has `recorded_version` only for legacy/current. Unknown reasons are:
+
+- `marker_absent`: retained history has no decision for this change ID, including
+  an older definition that has not reached the change point. A matching current
+  code fingerprint, build label, admission time or deployment time never upgrades
+  this to current.
+- `history_unavailable`: details are pruned, owning-instance/source provenance is
+  unavailable, the initial history/prefix is absent or incomplete, the stored
+  event format is not supported, or the bounded
+  inspection budget (1,000 version markers per run) is exceeded.
+- `history_invalid`: malformed decision data, invalid recorded range, or
+  conflicting decisions for the same change ID. Repeated decisions with the same
+  version remain valid when each recorded range is valid; range metadata may
+  differ across repeated historical calls. Marker integer fields use the signed
+  32-bit range.
+
+The v1 reader freezes its supported retained event-kind set; adding a runtime
+event enum case does not silently expand reader support.
+
+Marker readers must understand their runtime's retained format. Embedded marker
+fields are `change_id`, `version`, `min_supported`, `max_supported` and positive
+logical `sequence`; the range must contain the recorded version. The known
+producer-injected `task` snapshot is supported; unknown marker extensions require
+a reader upgrade and remain unavailable. Reused decisions must retain both
+`reused_from_run_id` and `reused_recorded_at`, or neither. The source ID is a
+nonempty bounded run identity; the clock is a valid RFC 3339 instant with an
+explicit offset and the runtime's supported precision (at most six fractional
+digits for embedded). Malformed or incomplete reuse
+provenance remains `history_invalid`; readers never rewrite retained clock strings.
+No application
+argument, output, memo or payload-store fetch is necessary for this query.
+
+## Response and paging
+
+The response shape is in
+[`resources/contracts/version-usage-v1.schema.json`](../resources/contracts/version-usage-v1.schema.json).
+`counts.total` equals legacy + current + unknown for the entire bounded open-run
+cohort, before the optional state filter and independently of page size/cursor.
+`runs` contains only the selected state when a state filter is present.
+
+Runs are ordered by byte-exact run ID ascending, with a strict keyset continuation.
+The first request captures the greatest matching open run ID as `through_run_id`;
+all continuations retain that upper bound and advance after the last returned
+ID. New admissions beyond that bound cannot appear on later pages. Cursors bind
+all identity/classification filters, and every page re-applies authorization and
+exact namespace predicates. Namespace changes cannot reuse a cursor.
+
+This is an explicitly **live**, read-only observation, not a durable fleet
+snapshot or deployment fence. Runs may close or record a marker between pages;
+counts are refreshed over the same upper-bounded cohort and may change. Returned
+IDs do not repeat within a cursor traversal, but a newly matching state behind
+the cursor requires a new traversal. Counts and a page are derived from the same
+per-run observations in each response. No response contains a `safe_to_remove`
+flag. Operators must keep unknowns visible and restart the query after draining
+or otherwise fencing admissions before making a rollout/removal decision.
+
+Example shape (the cursor is opaque):
+
+```json
+{
+  "schema_version": 1,
+  "namespace": "orders",
+  "workflow_type": "shipment",
+  "change_id": "carrier-api",
+  "min_version": 2,
+  "state": null,
+  "consistency": "live",
+  "through_run_id": "01J00000000000000000000002",
+  "counts": {"total": 3, "legacy": 1, "current": 1, "unknown": 1},
+  "runs": [{"instance_id": "shipment-1", "run_id": "01J00000000000000000000001", "status": "waiting", "state": "legacy", "recorded_version": 1, "unknown_reason": null}],
+  "limit": 1,
+  "returned_count": 1,
+  "has_more": true,
+  "next_cursor": "opaque-runtime-cursor"
+}
+```
+
+This draft is source-contract work only. Native endpoint/CLI/Waterline adoption
+and installed compatibility remain owned by their respective implementations.

resources/contracts/version-usage-v1.schema.json

@@ -0,0 +1,204 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "title": "Workflow open-run version usage v1",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "schema_version",
+    "namespace",
+    "workflow_type",
+    "change_id",
+    "min_version",
+    "state",
+    "consistency",
+    "through_run_id",
+    "counts",
+    "runs",
+    "limit",
+    "returned_count",
+    "has_more",
+    "next_cursor"
+  ],
+  "properties": {
+    "schema_version": {
+      "const": 1
+    },
+    "namespace": {
+      "type": [
+        "string",
+        "null"
+      ]
+    },
+    "workflow_type": {
+      "type": "string",
+      "minLength": 1
+    },
+    "change_id": {
+      "type": "string",
+      "minLength": 1
+    },
+    "min_version": {
+      "type": "integer",
+      "minimum": -2147483648,
+      "maximum": 2147483647
+    },
+    "state": {
+      "enum": [
+        null,
+        "legacy",
+        "current",
+        "unknown"
+      ]
+    },
+    "consistency": {
+      "const": "live"
+    },
+    "through_run_id": {
+      "type": [
+        "string",
+        "null"
+      ]
+    },
+    "counts": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": [
+        "total",
+        "legacy",
+        "current",
+        "unknown"
+      ],
+      "properties": {
+        "total": {
+          "type": "integer",
+          "minimum": 0
+        },
+        "legacy": {
+          "type": "integer",
+          "minimum": 0
+        },
+        "current": {
+          "type": "integer",
+          "minimum": 0
+        },
+        "unknown": {
+          "type": "integer",
+          "minimum": 0
+        }
+      }
+    },
+    "runs": {
+      "type": "array",
+      "maxItems": 200,
+      "items": {
+        "$ref": "#/$defs/run"
+      }
+    },
+    "limit": {
+      "type": "integer",
+      "minimum": 1,
+      "maximum": 200
+    },
+    "returned_count": {
+      "type": "integer",
+      "minimum": 0,
+      "maximum": 200
+    },
+    "has_more": {
+      "type": "boolean"
+    },
+    "next_cursor": {
+      "type": [
+        "string",
+        "null"
+      ]
+    }
+  },
+  "$defs": {
+    "run": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": [
+        "instance_id",
+        "run_id",
+        "status",
+        "state",
+        "recorded_version",
+        "unknown_reason"
+      ],
+      "properties": {
+        "instance_id": {
+          "type": "string"
+        },
+        "run_id": {
+          "type": "string"
+        },
+        "status": {
+          "enum": [
+            "pending",
+            "running",
+            "waiting"
+          ]
+        },
+        "state": {
+          "enum": [
+            "legacy",
+            "current",
+            "unknown"
+          ]
+        },
+        "recorded_version": {
+          "type": [
+            "integer",
+            "null"
+          ],
+          "minimum": -2147483648,
+          "maximum": 2147483647
+        },
+        "unknown_reason": {
+          "enum": [
+            null,
+            "marker_absent",
+            "history_unavailable",
+            "history_invalid"
+          ]
+        }
+      },
+      "allOf": [
+        {
+          "if": {
+            "properties": {
+              "state": {
+                "const": "unknown"
+              }
+            }
+          },
+          "then": {
+            "properties": {
+              "recorded_version": {
+                "type": "null"
+              },
+              "unknown_reason": {
+                "enum": [
+                  "marker_absent",
+                  "history_unavailable",
+                  "history_invalid"
+                ]
+              }
+            }
+          },
+          "else": {
+            "properties": {
+              "recorded_version": {
+                "type": "integer"
+              },
+              "unknown_reason": {
+                "type": "null"
+              }
+            }
+          }
+        }
+      ]
+    }
+  }
+}

scripts/ci/test-regression-corpus-policy.py

@@ -2141,6 +2141,35 @@ def test_core_replay_surfaces_require_growth_without_diff_keywords(self) -> None
                 )
                 target.write_text("<?php\nreturn 'base';\n", encoding="utf-8")
 
+    def test_untracked_history_observers_keep_the_existing_replay_guard(self) -> None:
+        for name in ("VersionUsageQuery", "VersionUsageDecision", "VersionUsageCursor", "AnotherHistoryHelper"):
+            with self.subTest(name=name):
+                source = self.root / f"src/V2/Support/{name}.php"
+                source.write_text(
+                    "<?php\nnamespace Workflow\\V2\\Support;\n"
+                    f"final class {name} {{ public function page($run) {{ return $run->last_history_sequence; }} }}",
+                    encoding="utf-8",
+                )
+                result = self.validate()
+                self.assertNotEqual(0, result.returncode, result.stdout)
+                self.assertIn("replay implementation changed but its corpus did not grow", result.stderr)
+                source.unlink()
+
+    def test_replacing_observation_with_transformed_raw_sql_still_requires_evidence(self) -> None:
+        source = self.root / "src/V2/Support/VersionUsageQuery.php"
+        header = "<?php\nnamespace Workflow\\V2\\Support;\nfinal class VersionUsageQuery { public function page($run) { "
+        source.write_text(header + "return $run->last_history_sequence; } }", encoding="utf-8")
+        self.commit_current_as_base()
+        for body in (
+            "$sql = base64_decode('aWQgSU5UTyBPVVRGSUxFICcvdG1wL3ZlcnNpb24tdXNhZ2Utd3JpdGUn'); return $run->selectRaw($sql)->get();",
+            "$sql = implode('', ['id IN', 'TO OUT', 'FILE name']); return $run->selectRaw($sql)->get();",
+        ):
+            with self.subTest(body=body):
+                source.write_text(header + body + " } }", encoding="utf-8")
+                result = self.validate()
+                self.assertNotEqual(0, result.returncode, result.stdout)
+                self.assertIn("replay implementation changed but its corpus did not grow", result.stderr)
+
     def test_task_discovery_helper_body_does_not_require_replay_growth(self) -> None:
         source = self.root / "src/V2/Support/DefaultWorkflowTaskBridge.php"
         original = """<?php

scripts/ci/validate-regression-corpus.py

@@ -2065,7 +2065,9 @@ def _guard_matches(
     )
     for path in matching:
         if path in untracked and (root / path).is_file():
-            diff += "\n" + (root / path).read_text(encoding="utf-8", errors="replace")
+            diff += "\n" + "\n".join(
+                "+" + line for line in (root / path).read_text(encoding="utf-8", errors="replace").splitlines()
+            )
     changed_content = "\n".join(
         line[1:]
         for line in diff.splitlines()