feat: reconcile granted ops run closes

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a06bfe-2a55-7ed3-bacd-879977b099bf

tests/test_ops_run_queue.py

@@ -118,7 +118,11 @@ def test_ops_run_to_dict_shape() -> None:
     row.source_id = "emit"
     row.triggering_event_id = "e1"
     row.approach_hint = None
+    row.harness_profile_ref = None
+    row.execution_refs = {}
+    row.repository_grant = None
     row.result = {}
+    row.close_intent_digest = None
     row.created_at = now
     row.updated_at = now
     d = ops_run_to_dict(row)
@@ -167,6 +171,33 @@ async def test_create_ops_run_inserts(monkeypatch: pytest.MonkeyPatch) -> None:
     session.execute.assert_awaited()
 
 
+@pytest.mark.asyncio
+async def test_create_ops_run_carries_repository_grant_separately(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from activity_core.ops_run_queue import create_ops_run_from_spec
+
+    monkeypatch.setenv("OPS_RUN_QUEUE_ENABLED", "true")
+    result = MagicMock()
+    result.scalar_one_or_none.return_value = uuid.uuid4()
+    session = AsyncMock()
+    session.execute = AsyncMock(return_value=result)
+    spec = TaskSpec(
+        title="bounded",
+        target_repo="activity-core",
+        activity_definition_id=str(uuid.uuid4()),
+        source_id="emit-bounded",
+        triggering_event_id="manual-1",
+    )
+    grant = _repository_grant()
+
+    await create_ops_run_from_spec(session, spec, repository_grant=grant)
+
+    statement = session.execute.await_args.args[0]
+    assert statement.compile().params["repository_grant"] == grant
+    assert statement.compile().params["execution_refs"] == {}
+
+
 @pytest.mark.asyncio
 async def test_claim_and_complete_roundtrip(monkeypatch: pytest.MonkeyPatch) -> None:
     """In-memory style: claim filters labels and complete transitions state."""
@@ -181,6 +212,8 @@ async def test_claim_and_complete_roundtrip(monkeypatch: pytest.MonkeyPatch) ->
     open_run.claim_owner = None
     open_run.lease_until = None
     open_run.result = {}
+    open_run.repository_grant = None
+    open_run.close_intent_digest = None
 
     session = AsyncMock()
 
@@ -234,6 +267,8 @@ async def test_complete_persists_only_normalized_glas_evidence() -> None:
     row.claim_owner = "worker-1"
     row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
     row.result = {}
+    row.repository_grant = None
+    row.close_intent_digest = None
     session = AsyncMock()
     session.get = AsyncMock(return_value=row)
 
@@ -279,6 +314,8 @@ async def test_fail_reopen_under_max_attempts(monkeypatch: pytest.MonkeyPatch) -
     row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
     row.attempt = 1
     row.result = {}
+    row.repository_grant = None
+    row.close_intent_digest = None
     session = AsyncMock()
     session.get = AsyncMock(return_value=row)
 
@@ -302,6 +339,8 @@ async def test_fail_permanent_at_max_attempts(monkeypatch: pytest.MonkeyPatch) -
     row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
     row.attempt = 3
     row.result = {}
+    row.repository_grant = None
+    row.close_intent_digest = None
     session = AsyncMock()
     session.get = AsyncMock(return_value=row)
 
@@ -322,6 +361,8 @@ async def test_fail_persists_redacted_failure_evidence() -> None:
     row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
     row.attempt = 1
     row.result = {}
+    row.repository_grant = None
+    row.close_intent_digest = None
     session = AsyncMock()
     session.get = AsyncMock(return_value=row)
 
@@ -390,6 +431,8 @@ async def test_complete_rejects_expired_lease_without_mutation() -> None:
     row.claim_owner = "worker-1"
     row.lease_until = now - timedelta(microseconds=1)
     row.result = {"before": True}
+    row.repository_grant = None
+    row.close_intent_digest = None
     session = AsyncMock()
     session.get = AsyncMock(return_value=row)
 
@@ -416,6 +459,8 @@ async def test_fail_rejects_wrong_owner_with_active_lease() -> None:
     row.claim_owner = "worker-1"
     row.lease_until = now + timedelta(minutes=1)
     row.result = {"before": True}
+    row.repository_grant = None
+    row.close_intent_digest = None
     session = AsyncMock()
     session.get = AsyncMock(return_value=row)
 
@@ -432,6 +477,155 @@ async def test_fail_rejects_wrong_owner_with_active_lease() -> None:
     assert row.result == {"before": True}
 
 
+@pytest.mark.asyncio
+async def test_terminal_complete_exact_repeat_reconciles_and_conflict_is_distinct() -> None:
+    from activity_core import ops_run_queue as oq
+
+    row = MagicMock()
+    row.state = "claimed"
+    row.claim_owner = "worker-1"
+    row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
+    row.result = {}
+    row.repository_grant = None
+    row.close_intent_digest = None
+    session = AsyncMock()
+    session.get = AsyncMock(return_value=row)
+
+    applied = await oq.close_ops_run(
+        session,
+        uuid.uuid4(),
+        worker_id="worker-1",
+        action="complete",
+        result={"ok": True, "path": "docs/result.md"},
+    )
+    digest = row.close_intent_digest
+    reconciled = await oq.close_ops_run(
+        session,
+        uuid.uuid4(),
+        worker_id="worker-1",
+        action="complete",
+        result={"path": "docs/result.md", "ok": True},
+    )
+    conflict = await oq.close_ops_run(
+        session,
+        uuid.uuid4(),
+        worker_id="worker-1",
+        action="complete",
+        result={"ok": True, "path": "docs/different.md"},
+    )
+    wrong_owner = await oq.close_ops_run(
+        session,
+        uuid.uuid4(),
+        worker_id="worker-2",
+        action="complete",
+        result={"ok": True, "path": "docs/result.md"},
+    )
+
+    assert applied.disposition == "applied"
+    assert reconciled.disposition == "reconciled"
+    assert conflict.disposition == "terminal_conflict"
+    assert wrong_owner.disposition == "wrong_owner"
+    assert row.state == "succeeded"
+    assert row.lease_until is None
+    assert row.close_intent_digest == digest
+
+
+@pytest.mark.asyncio
+async def test_close_reports_expired_lease_without_mutation() -> None:
+    from activity_core import ops_run_queue as oq
+
+    now = datetime(2026, 9, 4, 10, 0, tzinfo=timezone.utc)
+    row = MagicMock()
+    row.state = "claimed"
+    row.claim_owner = "worker-1"
+    row.lease_until = now
+    row.result = {"before": True}
+    row.repository_grant = None
+    row.close_intent_digest = None
+    session = AsyncMock()
+    session.get = AsyncMock(return_value=row)
+
+    with patch.object(oq, "_utcnow", return_value=now):
+        outcome = await oq.close_ops_run(
+            session,
+            uuid.uuid4(),
+            worker_id="worker-1",
+            action="complete",
+            result={"ok": True},
+        )
+
+    assert outcome.disposition == "expired_lease"
+    assert row.result == {"before": True}
+    assert row.close_intent_digest is None
+
+
+def _repository_grant() -> dict:
+    return {
+        "version": "1",
+        "allowed_paths": ["docs/", "README.md"],
+        "commit_count": {"min": 1, "max": 1},
+        "publish": False,
+    }
+
+
+def _accepted_repository_result() -> dict:
+    from activity_core.repository_grant import RepositoryGrant
+
+    grant = RepositoryGrant.model_validate(_repository_grant())
+    return {
+        "ok": True,
+        "repository_transaction": {
+            "transaction_id": "tx-1",
+            "repository_grant": {
+                "grant_id": grant.grant_id,
+                "acceptance_policy_id": grant.acceptance_policy_id,
+                "version": "1",
+            },
+            "acceptance": {
+                "accepted": True,
+                "policy_id": grant.acceptance_policy_id,
+                "head": "deadbeef",
+            },
+        },
+    }
+
+
+@pytest.mark.asyncio
+async def test_granted_complete_requires_matching_acceptance_identity() -> None:
+    from activity_core import ops_run_queue as oq
+
+    row = MagicMock()
+    row.state = "claimed"
+    row.claim_owner = "worker-1"
+    row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1)
+    row.result = {}
+    row.repository_grant = _repository_grant()
+    row.close_intent_digest = None
+    session = AsyncMock()
+    session.get = AsyncMock(return_value=row)
+
+    mismatch = _accepted_repository_result()
+    mismatch["repository_transaction"]["repository_grant"]["grant_id"] = "other"
+    refused = await oq.close_ops_run(
+        session,
+        uuid.uuid4(),
+        worker_id="worker-1",
+        action="complete",
+        result=mismatch,
+    )
+    applied = await oq.close_ops_run(
+        session,
+        uuid.uuid4(),
+        worker_id="worker-1",
+        action="complete",
+        result=_accepted_repository_result(),
+    )
+
+    assert refused.disposition == "evidence_conflict"
+    assert applied.disposition == "applied"
+    assert row.result["repository_transaction"]["transaction_id"] == "tx-1"
+
+
 def test_label_filter_any_vs_all() -> None:
     """Document labels_mode semantics used by claim_ops_runs."""
     row_labels = {"automated", "research-brief"}