feat(agent): declare workflow convergence criteria

agent/src/workflow/__init__.py

@@ -24,6 +24,8 @@
 )
 from .models import (
     AgentConfig,
+    Convergence,
+    ConvergenceEarlyExit,
     Hydration,
     Limits,
     ModelPreference,
@@ -47,6 +49,8 @@
 __all__ = [
     "STEP_HANDLERS",
     "AgentConfig",
+    "Convergence",
+    "ConvergenceEarlyExit",
     "Hydration",
     "Limits",
     "ModelPreference",

agent/src/workflow/models.py

@@ -44,6 +44,13 @@
 # 2026-06-08). First-party names: s3, comment, s3_and_comment.
 DeliverTarget = str
 TerminalOutcome = Literal["pr_url", "review_posted", "artifact", "comment"]
+ConvergenceMode = Literal["test_gated", "artifact_delivered", "human_approved", "review_submitted"]
+ConvergenceSensor = Literal["verify_build", "verify_lint"]
+# These are observable signals, unlike artifact-oriented TerminalOutcome values.
+# Mode/outcome coupling is enforced by the loader's JSON Schema pass.
+ConvergenceTerminalOutcome = Literal[
+    "pr_opened", "review_published", "artifact_delivered", "human_approved"
+]
 Status = Literal["draft", "validated", "production", "deprecated"]
 
 
@@ -151,6 +158,30 @@ class TerminalOutcomes(BaseModel):
     secondary: list[TerminalOutcome] = Field(default_factory=list)
 
 
+class ConvergenceEarlyExit(BaseModel):
+    """Explicit exceptions to the workflow's normal convergence path."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    allow_on_policy_deny: bool = False
+
+
+class Convergence(BaseModel):
+    """How a harness decides that this workflow is complete.
+
+    The v1 contract is declarative metadata for conformance tooling. Runtime
+    success inference continues to use the existing step and terminal-outcome
+    behavior until a later version explicitly adopts this contract.
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    mode: ConvergenceMode
+    required_sensors: list[ConvergenceSensor] | None = Field(default=None, min_length=1)
+    terminal_outcomes: list[ConvergenceTerminalOutcome]
+    early_exit: ConvergenceEarlyExit | None = None
+
+
 class Limits(BaseModel):
     """Workflow-level defaults; per-task / per-repo overrides still win."""
 
@@ -192,6 +223,7 @@ class Workflow(BaseModel):
     required_inputs: RequiredInputs | None = None
     steps: list[Step]
     terminal_outcomes: TerminalOutcomes
+    convergence: Convergence | None = None
     limits: Limits | None = None
     promotion_gate: PromotionGate | None = None
     status: Status

agent/tests/test_workflow_loader.py

@@ -11,8 +11,15 @@
 from typing import TYPE_CHECKING
 
 import pytest
+from pydantic import ValidationError
 
-from workflow import Workflow, WorkflowValidationError, load_workflow, load_workflow_file
+from workflow import (
+    Convergence,
+    Workflow,
+    WorkflowValidationError,
+    load_workflow,
+    load_workflow_file,
+)
 from workflow.loader import parse_workflow, validate_shape
 
 if TYPE_CHECKING:
@@ -45,6 +52,12 @@ def _valid_new_task() -> dict:
             {"kind": "ensure_pr", "name": "open_pr", "strategy": "create"},
         ],
         "terminal_outcomes": {"primary": "pr_url"},
+        "convergence": {
+            "mode": "test_gated",
+            "required_sensors": ["verify_build"],
+            "terminal_outcomes": ["pr_opened"],
+            "early_exit": {"allow_on_policy_deny": True},
+        },
         "limits": {"max_turns": 100},
         "promotion_gate": {"requires": ["tests:agent/new_task"]},
         "status": "production",
@@ -61,6 +74,12 @@ def test_parses_valid_coding_workflow(self):
         assert wf.steps[3].gate == "regression_only"
         assert wf.steps[4].strategy == "create"
         assert wf.terminal_outcomes.primary == "pr_url"
+        assert wf.convergence is not None
+        assert wf.convergence.mode == "test_gated"
+        assert wf.convergence.required_sensors == ["verify_build"]
+        assert wf.convergence.terminal_outcomes == ["pr_opened"]
+        assert wf.convergence.early_exit is not None
+        assert wf.convergence.early_exit.allow_on_policy_deny is True
 
     def test_unknown_top_level_field_rejected(self):
         body = _valid_new_task()
@@ -93,6 +112,119 @@ def test_requires_repo_false_forbids_clone_repo_via_schema(self):
         with pytest.raises(WorkflowValidationError):
             parse_workflow(body)
 
+    def test_convergence_is_optional_for_existing_workflows(self):
+        body = _valid_new_task()
+        del body["convergence"]
+        assert parse_workflow(body).convergence is None
+
+    def test_early_exit_policy_deny_defaults_false(self):
+        body = _valid_new_task()
+        body["convergence"]["early_exit"] = {}
+        workflow = parse_workflow(body)
+        assert workflow.convergence is not None
+        assert workflow.convergence.early_exit is not None
+        assert workflow.convergence.early_exit.allow_on_policy_deny is False
+
+    def test_parses_valid_human_approved_convergence(self):
+        body = _valid_new_task()
+        body["convergence"] = {
+            "mode": "human_approved",
+            "terminal_outcomes": ["human_approved"],
+        }
+        workflow = parse_workflow(body)
+        assert workflow.convergence is not None
+        assert workflow.convergence.mode == "human_approved"
+        assert workflow.convergence.required_sensors is None
+        assert workflow.convergence.terminal_outcomes == ["human_approved"]
+
+    def test_model_rejects_explicit_empty_required_sensors(self):
+        with pytest.raises(ValidationError, match="required_sensors"):
+            Convergence(
+                mode="review_submitted",
+                required_sensors=[],
+                terminal_outcomes=["review_published"],
+            )
+
+    @pytest.mark.parametrize(
+        ("convergence", "error_path"),
+        [
+            (
+                {
+                    "mode": "fixed_iterations",
+                    "terminal_outcomes": ["pr_opened"],
+                },
+                "convergence/mode",
+            ),
+            (
+                {
+                    "mode": "test_gated",
+                    "terminal_outcomes": ["pr_opened"],
+                },
+                "convergence",
+            ),
+            (
+                {
+                    "mode": "test_gated",
+                    "required_sensors": [],
+                    "terminal_outcomes": ["pr_opened"],
+                },
+                "convergence/required_sensors",
+            ),
+            (
+                {
+                    "mode": "test_gated",
+                    "required_sensors": ["verify_tests"],
+                    "terminal_outcomes": ["pr_opened"],
+                },
+                "convergence/required_sensors/0",
+            ),
+            (
+                {
+                    "mode": "review_submitted",
+                    "terminal_outcomes": ["review_published"],
+                    "early_exit": {"ignore_failures": True},
+                },
+                "convergence/early_exit",
+            ),
+            (
+                {
+                    "mode": "review_submitted",
+                    "terminal_outcomes": ["pr_opened"],
+                },
+                "convergence/terminal_outcomes",
+            ),
+            (
+                {
+                    "mode": "artifact_delivered",
+                    "terminal_outcomes": ["pr_opened"],
+                },
+                "convergence/terminal_outcomes",
+            ),
+            (
+                {
+                    "mode": "human_approved",
+                    "terminal_outcomes": ["pr_opened"],
+                },
+                "convergence/terminal_outcomes",
+            ),
+        ],
+    )
+    def test_invalid_convergence_rejected_with_clear_path(self, convergence, error_path):
+        body = _valid_new_task()
+        body["convergence"] = convergence
+        with pytest.raises(WorkflowValidationError) as exc:
+            parse_workflow(body)
+        assert f"{error_path}:" in str(exc.value)
+
+    @pytest.mark.parametrize("sensor", ["verify_build", "verify_lint"])
+    def test_required_sensor_must_have_matching_step(self, sensor):
+        body = _valid_new_task()
+        body["convergence"]["required_sensors"] = [sensor]
+        body["steps"] = [step for step in body["steps"] if step["kind"] != sensor]
+        with pytest.raises(WorkflowValidationError) as exc:
+            parse_workflow(body)
+        assert "steps:" in str(exc.value)
+
 
 class TestRequiresRepoDefault:
     def test_coding_defaults_true(self):
@@ -113,6 +245,10 @@ def test_knowledge_defaults_false(self):
             {"kind": "deliver_artifact", "target": "s3_and_comment"},
         ]
         body["terminal_outcomes"] = {"primary": "artifact"}
+        body["convergence"] = {
+            "mode": "artifact_delivered",
+            "terminal_outcomes": ["artifact_delivered"],
+        }
         assert parse_workflow(body).resolved_requires_repo is False
 
     def test_explicit_value_overrides_domain_default(self):
@@ -175,6 +311,20 @@ def test_invalid_yaml_rejected(self, tmp_path: Path):
 
 
 class TestLoadWorkflow:
+    @pytest.mark.parametrize(
+        ("workflow_id", "mode", "outcome"),
+        [
+            ("coding/new-task-v1", "test_gated", "pr_opened"),
+            ("coding/pr-iteration-v1", "test_gated", "pr_opened"),
+            ("coding/pr-review-v1", "review_submitted", "review_published"),
+        ],
+    )
+    def test_shipped_coding_workflow_declares_convergence(self, workflow_id, mode, outcome):
+        workflow = load_workflow(workflow_id)
+        assert workflow.convergence is not None
+        assert workflow.convergence.mode == mode
+        assert workflow.convergence.terminal_outcomes == [outcome]
+
     def test_missing_id_raises(self):
         with pytest.raises(WorkflowValidationError, match="not found"):
             load_workflow("coding/does-not-exist-v9")

agent/workflows/coding/new-task-v1.yaml

@@ -46,6 +46,10 @@ steps:
   - { kind: ensure_pr,       name: open_pr, strategy: create }
 terminal_outcomes:
   primary: pr_url
+convergence:
+  mode: test_gated
+  required_sensors: [verify_build]
+  terminal_outcomes: [pr_opened]
 limits:
   max_turns: 100
 promotion_gate:

agent/workflows/coding/pr-iteration-v1.yaml

@@ -43,6 +43,10 @@ steps:
   - { kind: ensure_pr,       name: resolve_pr, strategy: push_resolve }
 terminal_outcomes:
   primary: pr_url
+convergence:
+  mode: test_gated
+  required_sensors: [verify_build]
+  terminal_outcomes: [pr_opened]
 limits:
   max_turns: 100
 promotion_gate:

agent/workflows/coding/pr-review-v1.yaml

@@ -47,6 +47,9 @@ steps:
   - { kind: ensure_pr,       name: resolve_pr, strategy: resolve }
 terminal_outcomes:
   primary: pr_url
+convergence:
+  mode: review_submitted
+  terminal_outcomes: [review_published]
 limits:
   max_turns: 100
 promotion_gate: