diff --git a/README.md b/README.md index f84ce86..4af34c2 100644 --- a/README.md +++ b/README.md @@ -90,10 +90,11 @@ Each `run` resolves the agent's tool profile + budget from the target repo's instance manifest (default `green-commit-only`), loads a persona bundle (`kaizen-agentic schedule prepare`), runs a bounded agentic session (cwd-pinned, hard allow-list, never pushes), verifies a local commit, -records metrics, and posts a State Hub progress event. Local task files with a -v1 `repository_grant` receive full repository acceptance and mandatory durable -external metrics; grant-absent compatibility runs retain repository-local -`.kaizen/metrics` during migration. +records metrics, and posts a State Hub progress event. Local task files and +profiled Activity Core rows with a v1 `repository_grant` receive full repository +acceptance and mandatory durable external metrics; profiled terminal closes +are outboxed before delivery. Grant-absent compatibility runs retain +repository-local `.kaizen/metrics` during migration. When invoked through `glas-harness`, the explicit versioned Glas profile may override model, tool profile, and budget for that bounded run. Workforce and activity consumers should reference the Glas profile rather than encode these @@ -103,6 +104,6 @@ Instance manifest contract: [docs/instance-manifest.md](docs/instance-manifest.m Example: [examples/schedule.harness.yml](examples/schedule.harness.yml). Repository mutation authority: [docs/repository-grant.md](docs/repository-grant.md). Granted-run metrics: [docs/external-metrics.md](docs/external-metrics.md). -Durable result-close design: [docs/close-evidence-outbox.md](docs/close-evidence-outbox.md). +Durable result-close operation: [docs/close-evidence-outbox.md](docs/close-evidence-outbox.md). Tests: `PYTHONPATH=".:$HOME/llm-connect" python3 -m pytest tests/ -q` diff --git a/SCOPE.md b/SCOPE.md index e2447c1..dbfb2e9 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -19,8 +19,9 @@ retains a legacy set of local task executors for unattended repository work. fails them with normalized result data. - A queued `harness_profile_ref` is authoritative. The worker carries its allowlisted attribution refs into a Glas `ExecutionRequest`, uses governed - actor `agt`, returns the complete `GatewayResult`, and never falls back to a - legacy approach after profile refusal or failure. + actor `agt`, retains bounded execution evidence, and never falls back to a + legacy approach after profile refusal or failure. A separately typed queued + repository grant is parsed strictly and cannot cross a profile-absent route. - Profile-absent rows use the coexistence registry in `approaches.py`. Dispatch requires an explicit, non-expired `AGENT_HARNESS_LEGACY_APPROACHES_UNTIL` date; unmatched or disabled routes @@ -43,7 +44,8 @@ retains a legacy set of local task executors for unattended repository work. reported to State Hub. - Versioned profiled rows delegate profile, rein, model, tool, sandbox, and teardown decisions to Glas/sand-boxer. `rein-aharness` owns the queue - adapter, not the meta-framework or sandbox implementation. + adapter, repository transaction/acceptance, external metrics, and durable + terminal-close delivery—not the meta-framework or sandbox implementation. - Legacy structured executors produce Freedom Intelligence daily research briefs, Binky daily/weekly briefs, deterministic mailbox scans, and mailbox triage. The LLM-backed structured paths use the llm-connect HTTP service; @@ -67,6 +69,10 @@ retains a legacy set of local task executors for unattended repository work. Granted runs instead require private, durable external metrics with a projection descriptor so accepted checkouts remain clean. State Hub progress/tool/token events and task closing remain best-effort. +- Profiled terminal closes are written to a private outbox before delivery. + Pending close-only work replays before another claim; exact Activity Core + repeats reconcile, permanent machine-distinct conflicts quarantine, and + operator status/replay commands expose unresolved evidence. - llm-connect HTTP errors retain only bounded, allowlisted provider diagnosis. ### Packaging and deployment @@ -94,10 +100,10 @@ retains a legacy set of local task executors for unattended repository work. the host checkout. They are not an OS-level sandbox and do not prove that no push/network activity occurred after the run. Unattended mutation now takes a process-safe repository lock (ADR-002 / T02). -- Explicitly granted local task files have full repository acceptance and - clean external metrics. Grant-absent compatibility and profiled queue runs - still use their prior result contracts because Activity Core carries no - authoritative repository grant. Durable queue-close replay is also not live. +- Explicitly granted local task files and profiled Activity Core rows have full + repository acceptance and clean external metrics. Source compatibility is + pinned to Activity Core `b63131e` / schema migration `0010`; production must + deploy that schema/API contract before a granted definition is enabled. - The worker is sequential and repository mapping is host configuration. There is no multi-worker repository lease, per-tenant process isolation, generic credential broker, or tenant onboarding API. diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index a46a26d..65ab2e4 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -8,6 +8,7 @@ | Kind | ID | Status | Lane | Source | | --- | --- | --- | --- | --- | +| workplan | HARNESS-WP-ADHOC-2026-09-04 | finished | — | workplans/ADHOC-2026-09-04.md | | workplan | REINAH-WP-0001 | done | — | workplans/REINAH-WP-0001-harness-foundation.md | | workplan | REINAH-WP-0002 | finished | — | workplans/REINAH-WP-0002-rename-and-glas-harness-alignment.md | | workplan | REINAH-WP-0003 | active | — | workplans/REINAH-WP-0003-governed-runtime-integrity.md | @@ -15,6 +16,7 @@ | workplan | REINAH-WP-0005 | finished | — | workplans/REINAH-WP-0005-ops-run-claim-loop.md | | workplan | REINAH-WP-0006 | finished | — | workplans/REINAH-WP-0006-binky-weekly-review-approach.md | | workplan | REINAH-WP-0007 | finished | — | workplans/REINAH-WP-0007-glas-profiled-ops-runs.md | +| task | HARNESS-WP-ADHOC-2026-09-04-T01 | done | — | workplans/ADHOC-2026-09-04.md | | task | REINAH-WP-0001-T01 | done | — | workplans/REINAH-WP-0001-harness-foundation.md | | task | REINAH-WP-0001-T02 | done | — | workplans/REINAH-WP-0001-harness-foundation.md | | task | REINAH-WP-0001-T03 | done | — | workplans/REINAH-WP-0001-harness-foundation.md | @@ -28,7 +30,7 @@ | task | REINAH-WP-0002-T04 | done | — | workplans/REINAH-WP-0002-rename-and-glas-harness-alignment.md | | task | REINAH-WP-0003-T01 | done | — | workplans/REINAH-WP-0003-governed-runtime-integrity.md | | task | REINAH-WP-0003-T02 | done | — | workplans/REINAH-WP-0003-governed-runtime-integrity.md | -| task | REINAH-WP-0003-T03 | wait | — | workplans/REINAH-WP-0003-governed-runtime-integrity.md | +| task | REINAH-WP-0003-T03 | done | — | workplans/REINAH-WP-0003-governed-runtime-integrity.md | | task | REINAH-WP-0003-T04 | wait | — | workplans/REINAH-WP-0003-governed-runtime-integrity.md | | task | REINAH-WP-0003-T05 | progress | — | workplans/REINAH-WP-0003-governed-runtime-integrity.md | | task | REINAH-WP-0003-T06 | wait | — | workplans/REINAH-WP-0003-governed-runtime-integrity.md | diff --git a/docs/architecture.md b/docs/architecture.md index 8cf0eb1..c77ed5d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -34,11 +34,12 @@ |---|---|---|---| | Instance manifest | `manifest.py` | extends `.kaizen/schedule.yml`; `validate` CLI | unchanged contract; tenant onboarding | | Tool profiles | `profiles.py` | `green-commit-only`, `blue-mail-triage` registry | additional named profiles as needed | -| Task intake | `intake.py` + `taskspec.py` | issue-core GET/PATCH poll+claim; JSON task-file for local dev | NATS when activity-core migrates | +| Task intake | `ops_run_client.py` + `taskspec.py` | Activity Core claim/lease/close; JSON task-file for local dev; issue-core compatibility | event-driven wakeup if needed | | Persona | `persona.py` | `kaizen-agentic schedule prepare` (ADR-005) | unchanged, plus phase-memory profile hook | | Session | `adapter.py` | `AgenticClaudeCodeAdapter` (cwd-pinned, profile allow-list) | + hosted adapters | -| Orchestration | `runner.py` | profile/budget → transaction → session → grant acceptance → metrics → hub | activity-core intake | +| Orchestration | `runner.py` + `claim_loop.py` | profile → transaction → session → grant acceptance → metrics → durable close | readiness and recovery drills | | Metrics | `metrics.py` | compatibility `.kaizen/metrics`; durable external ledger for granted runs | kaizen-owned projection | +| Close evidence | `close_outbox.py` | enqueue-before-close; exact replay; conflict quarantine | operational retention policy | | Mail lane | `mailscan.py` | deterministic credentialed pre-step outside the session | pattern generalizes to other credentialed pre-steps | | Hub reporting | `hub.py` | REST progress event + task close + token event | unchanged | diff --git a/docs/close-evidence-outbox.md b/docs/close-evidence-outbox.md index 1b46d0d..5f00df2 100644 --- a/docs/close-evidence-outbox.md +++ b/docs/close-evidence-outbox.md @@ -1,6 +1,6 @@ # Close-evidence outbox -Status: **durable core implemented, not connected to the live claim loop**. +Status: **active for profiled Activity Core terminal closes**. The outbox preserves one bounded Activity Core completion or failure intent after repository work. Replaying an outbox entry calls only the queue close @@ -21,9 +21,10 @@ outside target checkouts. Outbox directories are mode `0700`; lock and JSON files are mode `0600`. An entry is keyed by the digest of the bounded Activity Core `run_id` and -repository `transaction_id`. Re-enqueuing the identical intent returns its -existing pending or delivered state. Reusing that identity for a different -action, result, error, or reopen decision is a conflict and is refused. +repository `transaction_id` and retains the original worker identity. +Re-enqueuing the identical intent returns its existing pending or delivered +state. Reusing that identity for a different worker, action, result, error, or +reopen decision is a conflict and is refused. The v1 close intent contains: @@ -41,7 +42,8 @@ Enqueue and retry-state updates use a same-directory temporary file, file `fsync`, atomic replace, and directory `fsync` under a process-safe `flock`. The retry attempt is durable before the delivery callback begins. -Normal callback return marks the entry delivered and atomically moves it from +Normal callback return, including Activity Core `close_disposition=applied` or +`reconciled`, marks the entry delivered and atomically moves it from `pending/` to `delivered/`. An ordinary exception retains it in `pending/` with only the exception class and a generic failure marker; exception text is not persisted. Process interrupts propagate; the already-recorded attempt remains @@ -49,21 +51,29 @@ pending. Invalid, oversized, or directory/state-mismatched records are moved intact to `quarantine/` with a bounded reason sidecar and are never delivered automatically. -A network timeout may occur after Activity Core accepted a close. Retrying the -same close intent can therefore repeat the queue mutation request, but cannot -repeat repository workload. Activity Core remains authoritative for terminal -row state and must provide idempotent or reconcilable close semantics. +Activity Core refusal codes `not_found`, `wrong_owner`, `expired_lease`, +`state_conflict`, `evidence_conflict`, and `terminal_conflict` are conclusive, +not transient. Such entries move to quarantine rather than retrying forever. +An unclassified HTTP failure or transport outage remains pending. -## Activation gates +A network timeout may occur after Activity Core accepted a close. The worker +replays pending close-only work before it claims another row. Retrying the same +intent can repeat the queue request, but cannot repeat repository workload; +Activity Core reconciles the exact normalized terminal repeat. -The live claim loop still returns an in-memory failure when close delivery -fails. Connecting it to this outbox requires: +## Operator controls -1. the lease-bound repository transaction and accepted-result envelope; -2. enqueue-before-close using the same run and transaction identities; -3. Activity Core review of repeat close behavior for already-terminal rows; -4. startup/periodic replay that never calls workload code; and -5. operator status, quarantine inspection, and replay controls. +```bash +rein-aharness close-outbox status +rein-aharness close-outbox replay +rein-aharness close-outbox replay --limit 10 +``` -Until those gates are implemented and reviewed, no production close behavior -changes. +`status` and `replay` return non-zero while pending or quarantined material +exists, making them suitable for readiness/recovery checks. Quarantine content +and its bounded reason sidecar remain under the private state path for explicit +operator inspection; it is never silently deleted or treated as delivered. + +Profile-absent compatibility closes are not yet outboxed. Granted rows without +an authoritative `harness_profile_ref` are refused before execution so they +cannot cross that weaker path. diff --git a/docs/legacy-runtime-inventory.md b/docs/legacy-runtime-inventory.md index f4a617c..9d3f50e 100644 --- a/docs/legacy-runtime-inventory.md +++ b/docs/legacy-runtime-inventory.md @@ -42,10 +42,10 @@ must never fall back to either. An unmatched route fails terminally. Most matched executor failures set `reopen=true`; an unexpected exception also reopens. Activity Core applies its -attempt limit. The harness now records a repository transaction identity, but -Activity Core does not yet carry an accepted repository grant or reconcile a -response-lost terminal close, so a retry after ambiguous completion remains a -T03 gate. +attempt limit. Grant-bearing work now requires an authoritative profile and +therefore cannot enter this registry. Profiled terminal closes use durable +response-loss reconciliation; legacy close delivery remains a T04 migration +constraint rather than a grant-bearing path. ## Scheduled definitions diff --git a/docs/ops-run-claim-loop.md b/docs/ops-run-claim-loop.md index 2a60b40..22d00a0 100644 --- a/docs/ops-run-claim-loop.md +++ b/docs/ops-run-claim-loop.md @@ -9,7 +9,7 @@ Does **not** require issue-core or Forgejo for FI / Binky-style ops. ```text activity-core emit_tasks - → INSERT ops_run (open, optional harness_profile_ref) + activity_task_spawn + → INSERT ops_run (open, optional profile + repository_grant) + task_spawn │ ▼ rein-aharness claim-loop @@ -17,15 +17,18 @@ rein-aharness claim-loop harness_profile_ref present? ─ yes → Glas ExecutionRequest → gateway │ no └─→ select_approach(labels, definition, hint) → legacy executor - POST complete | fail(+reopen) + validate granted repository result + persist external metrics + enqueue exact terminal close → POST complete | fail + replay pending close evidence before another claim ``` `harness_profile_ref` is authoritative under ACT-ADR-006. A profiled row never consults `approach_hint`, `select_approach`, or a legacy executor, including -after profile resolution refusal. The claim worker sends the complete -`GatewayResult` to Activity Core; Activity Core persists its allowlisted -`ExecutionEvidence` and drops direct tool output/error. Profile-absent rows keep -the legacy coexistence path until the migration flag is enabled upstream. +after profile resolution refusal. The claim worker sends bounded +`ExecutionEvidence`; direct tool output/error and unknown provider fields are +not persisted in its close outbox. A separately typed grant activates commit, +path, clean-state, metadata, and remote-ref acceptance plus external metrics. +Profile-absent rows keep the legacy coexistence path, but cannot carry a grant. ## Environment @@ -87,6 +90,10 @@ rein-aharness poll --source=ops-run rein-aharness claim-loop rein-aharness claim-loop --once --verbose +# Required close evidence and recovery +rein-aharness close-outbox status +rein-aharness close-outbox replay + # Dry-run: claim then fail+reopen without domain work rein-aharness poll --source=ops-run --dry-run ``` diff --git a/docs/repository-grant.md b/docs/repository-grant.md index 837d1ec..07acdb9 100644 --- a/docs/repository-grant.md +++ b/docs/repository-grant.md @@ -1,7 +1,7 @@ # Repository grant contract -Status: **v1 enabled for local `TaskSpec` files; queued/profiled carriage is -not yet available**. +Status: **v1 enabled for local `TaskSpec` files and profiled Activity Core +rows**. `repository_grant` is the explicit authority envelope for a bounded local Git mutation. It is separate from task prose, labels, organizational attribution, @@ -53,16 +53,18 @@ Granted runs require durable external metrics and refuse `--no-metrics`. Their metrics are written outside the checkout only after acceptance, so a successful run remains clean. See [external-metrics.md](external-metrics.md). -Activity Core and issue-core adapters do not populate `repository_grant`. -Queued/profiled runs therefore cannot claim repository acceptance under this -contract. Existing grant-absent direct and compatibility runs retain their -legacy `HEAD`-changed behavior during migration. +Activity Core revision `b63131e` added strict definition admission and separate +queue carriage. The claim client parses that exact v1 field, refuses malformed +authority, and requires `harness_profile_ref` whenever a queued grant is +present. Profiled success applies the same repository acceptance policy and +external metrics contract as a local granted run. -Queued/profiled execution may use this contract only after: - -1. an authoritative Activity Core/profile field carries the reviewed grant; -2. required close evidence durably records the grant, transaction, and accepted - result identities. +Before sending a profiled terminal result, the worker durably records the exact +run, transaction, worker, action, and bounded result. Response-lost closes are +replayed without repeating workload code and Activity Core distinguishes exact +reconciliation from terminal, evidence, owner, state, and lease conflicts. +Grant-absent compatibility runs retain their legacy result behavior during +migration. Task descriptions, labels, `execution_refs`, consuming-repo defaults, and profile absence must never synthesize a grant. diff --git a/docs/task-intake.md b/docs/task-intake.md index fa912f4..2298269 100644 --- a/docs/task-intake.md +++ b/docs/task-intake.md @@ -32,10 +32,10 @@ rein-aharness claim-loop `TaskExecutorWorkflow` in activity-core stays a stub; execution lives here. -Activity Core does not yet emit the versioned `repository_grant` required by -the lease-bound transaction path. The harness must not infer one from labels, -task prose, `execution_refs`, or target-repo lookup. See -**`docs/repository-grant.md`** for the parsed-but-not-enabled v1 contract. +Activity Core revision `b63131e` emits a separately typed versioned +`repository_grant`. The harness parses it strictly, requires an authoritative +profile for queued grants, and never infers authority from labels, task prose, +`execution_refs`, or target-repo lookup. See **`docs/repository-grant.md`**. --- diff --git a/rein_aharness/claim_loop.py b/rein_aharness/claim_loop.py index 27b51dd..5fb7800 100644 --- a/rein_aharness/claim_loop.py +++ b/rein_aharness/claim_loop.py @@ -17,12 +17,19 @@ import time from dataclasses import dataclass, field from typing import Any +from rein_aharness import metrics from rein_aharness.approaches import ( APPROACH_UNMATCHED, ApproachResult, execute_approach, select_approach, ) +from rein_aharness.close_outbox import ( + CloseOutbox, + CloseRequest, + PermanentCloseDeliveryError, + ReplayReport, +) from rein_aharness.execution_cancel import ( ExecutionCancel, ExecutionCancelled, @@ -32,11 +39,13 @@ from rein_aharness.glas_execution import ( GLAS_APPROACH, GlasExecutionError, execute_profiled_run, + normalise_execution_evidence_for_close, ) from rein_aharness.lease_monitor import LeaseLoss, LeaseMonitor from rein_aharness.repository_transaction import ( DirtyRepositoryError, GitRepositoryError, + RepositoryAcceptanceError, RepositoryBusyError, RepositoryTransaction, RepositoryTransactionError, @@ -52,6 +61,17 @@ from rein_aharness.ops_run_client import ( logger = logging.getLogger("rein_aharness.claim_loop") +_PERMANENT_CLOSE_CODES = frozenset( + { + "not_found", + "wrong_owner", + "expired_lease", + "state_conflict", + "terminal_conflict", + "evidence_conflict", + } +) + @dataclass class ProcessResult: @@ -180,16 +200,97 @@ def _cancelled_result( return None +def _deliver_close(client: ActivityCoreOpsClient, request: CloseRequest) -> OpsRun: + if request.worker_id != client.config.worker_id: + raise PermanentCloseDeliveryError("outbox worker identity no longer matches") + try: + if request.action == "complete": + return client.complete(request.run_id, result=request.result) + return client.fail( + request.run_id, + error=request.error, + reopen=request.reopen, + result=request.result, + ) + except OpsRunError as exc: + if exc.code in _PERMANENT_CLOSE_CODES: + raise PermanentCloseDeliveryError( + f"Activity Core refused close with {exc.code}" + ) from exc + raise + + +def replay_close_outbox( + client: ActivityCoreOpsClient, + outbox: CloseOutbox | None = None, + *, + limit: int = 100, +) -> ReplayReport: + """Replay close-only work; this function never dispatches a workload.""" + store = outbox or CloseOutbox() + return store.replay(lambda request: _deliver_close(client, request), limit=limit) + + +def _durable_close( + client: ActivityCoreOpsClient, + outbox: CloseOutbox, + request: CloseRequest, +) -> tuple[bool, str | None, str]: + """Enqueue before delivery and report the durable outcome for this entry.""" + receipt = outbox.enqueue(request) + delivered: dict[str, OpsRun] = {} + + def deliver(item: CloseRequest) -> None: + delivered[item.entry_id] = _deliver_close(client, item) + + report = outbox.replay(deliver) + state = outbox.entry_state(receipt.entry_id) + if state != "delivered": + reason = ( + "close evidence quarantined" + if state == "quarantined" + else "close evidence remains pending" + ) + return False, None, reason + row = delivered.get(receipt.entry_id) + if row is not None: + return True, row.state, "" + expected_state = "succeeded" if request.action == "complete" else "failed" + if report.delivered: + logger.info("replayed %s close evidence entries", report.delivered) + return True, expected_state, "" + + +def _bounded_reason(value: str, *, default: str) -> str: + cleaned = " ".join(str(value).split()) + return cleaned[:500] or default + + def process_one( client: ActivityCoreOpsClient | None = None, *, report_to_hub: bool = True, commit: bool = True, dry_run: bool = False, + outbox: CloseOutbox | None = None, ) -> ProcessResult: """Claim at most one ops_run, execute approach, complete or fail.""" client = client or ActivityCoreOpsClient() cfg = client.config + outbox = outbox or CloseOutbox() + + replay = replay_close_outbox(client, outbox) + if replay.remaining: + return ProcessResult( + claimed=False, + retry_full_interval=True, + reason=( + "required Activity Core close evidence remains pending " + f"(attempted={replay.attempted} remaining={replay.remaining})" + ), + ) + if replay.quarantined: + logger.error("quarantined %s close evidence entries", replay.quarantined) try: claimed = client.claim(limit=1) @@ -244,6 +345,36 @@ def process_one( client, run, report_to_hub=report_to_hub, + outbox=outbox, + ) + + if run.repository_grant is not None: + reason = ( + "refused: repository_grant requires an authoritative " + "harness_profile_ref" + ) + try: + out = client.fail( + run.id, + error=reason, + reopen=False, + result={"ok": False, "approach": approach, "reason": reason}, + ) + except OpsRunError as exc: + return ProcessResult( + claimed=True, + run_id=run.id, + approach=approach, + ok=False, + reason=f"grant-route refusal close failed: {exc}", + ) + return ProcessResult( + claimed=True, + run_id=run.id, + approach=approach, + ok=False, + reason=reason, + ops_state=out.state, ) cancel = ExecutionCancel() @@ -340,61 +471,213 @@ def _process_profiled_run( run: OpsRun, *, report_to_hub: bool, + outbox: CloseOutbox, ) -> ProcessResult: - """Execute a profiled row without consulting or falling back to legacy routing.""" + """Keep the claim heartbeat alive through evidence persistence and close.""" cancel = ExecutionCancel() hb = _Heartbeat(client, run.id, client.config.lease_seconds, cancel=cancel) hb.start() - execution_error: str | None = None - execution_reason = "" - execution_reopen = False - gateway_result: dict[str, Any] | None = None - tx_evidence: dict[str, Any] | None = None _set_active_run_cancel(cancel) try: - try: - target = resolve_ops_target(run, client.config) - with using_cancel(cancel): - with RepositoryTransaction(target, correlation_id=run.id) as tx: - gateway_result = execute_profiled_run( - run, - client.config, - report_to_hub=report_to_hub, - cancel=cancel, - ) - tx_evidence = tx.evidence() - except ExecutionCancelled as exc: - execution_error = type(exc).__name__ - execution_reason = str(exc) - except DirtyRepositoryError as exc: - execution_error = type(exc).__name__ - execution_reason = f"refused: {exc}" - except RepositoryBusyError as exc: - execution_error = type(exc).__name__ - execution_reason = f"refused: {exc}" - execution_reopen = True - except (GitRepositoryError, RepositoryTransactionError, TaskSpecError) as exc: - execution_error = type(exc).__name__ - execution_reason = f"refused: {exc}" - except GlasExecutionError as exc: - execution_error = type(exc).__name__ - execution_reason = str(exc) + return _process_profiled_run_active( + client, + run, + report_to_hub=report_to_hub, + outbox=outbox, + cancel=cancel, + hb=hb, + ) finally: hb.stop() _set_active_run_cancel(None) + +def _process_profiled_run_active( + client: ActivityCoreOpsClient, + run: OpsRun, + *, + report_to_hub: bool, + outbox: CloseOutbox, + cancel: ExecutionCancel, + hb: _Heartbeat, +) -> ProcessResult: + """Execute a profiled row without consulting or falling back to legacy routing.""" + execution_error: str | None = None + execution_reason = "" + execution_reopen = False + gateway_result: dict[str, Any] | None = None + safe_evidence: dict[str, Any] = {} + execution_ok = False + tx: RepositoryTransaction | None = None + tx_evidence: dict[str, Any] | None = None + try: + target = resolve_ops_target(run, client.config) + with using_cancel(cancel): + with RepositoryTransaction(target, correlation_id=run.id) as active_tx: + tx = active_tx + gateway_result = execute_profiled_run( + run, + client.config, + report_to_hub=report_to_hub, + cancel=cancel, + ) + safe_evidence = normalise_execution_evidence_for_close( + gateway_result["evidence"] + ) + execution_ok = gateway_result["ok"] + execution_reason = _bounded_reason( + str( + safe_evidence.get("error") + or safe_evidence.get("outcome") + or "" + ), + default="Glas execution failed", + ) + if run.repository_grant is not None: + try: + active_tx.validate_acceptance( + run.repository_grant.acceptance_policy() + ) + except RepositoryAcceptanceError as exc: + if execution_ok or exc.code != "head-unchanged": + execution_ok = False + execution_error = type(exc).__name__ + execution_reason = _bounded_reason( + str(exc), + default="repository acceptance failed", + ) + except ExecutionCancelled as exc: + execution_error = type(exc).__name__ + execution_reason = _bounded_reason(str(exc), default="execution cancelled") + except DirtyRepositoryError as exc: + execution_error = type(exc).__name__ + execution_reason = _bounded_reason( + f"refused: {exc}", default="dirty repository refused" + ) + except RepositoryBusyError as exc: + execution_error = type(exc).__name__ + execution_reason = _bounded_reason( + f"refused: {exc}", default="busy repository refused" + ) + execution_reopen = True + except (GitRepositoryError, RepositoryTransactionError, TaskSpecError) as exc: + execution_error = type(exc).__name__ + execution_reason = _bounded_reason( + f"refused: {exc}", default="repository transaction refused" + ) + except GlasExecutionError: + execution_error = "GlasExecutionError" + execution_reason = "profiled execution failed (GlasExecutionError)" + + if tx is not None and tx.baseline is not None: + tx_evidence = tx.evidence() + if run.repository_grant is not None: + tx_evidence["repository_grant"] = run.repository_grant.evidence() + + if run.repository_grant is not None and tx is not None: + try: + accepted = tx.acceptance + commit_sha = safe_evidence.get("commit_sha") + metrics.record_external_execution( + tx.repo, + "rein-aharness", + success=( + execution_error is None + and execution_ok + and not cancel.cancelled + and not hb.monitor.lost + ), + committed=bool(accepted or commit_sha), + head_after=( + accepted.head + if accepted is not None + else commit_sha if isinstance(commit_sha, str) else None + ), + reason=execution_reason or None, + metadata={ + "ops_run_id": run.id, + "harness_profile_ref": run.harness_profile_ref, + "repository_grant_id": run.repository_grant.grant_id, + }, + session_id=tx.transaction_id, + ) + assert tx_evidence is not None + tx_evidence["metrics"] = { + "storage": "external", + "session_id": tx.transaction_id, + "projection_ready": True, + } + except OSError: + execution_ok = False + execution_error = "ExternalMetricsError" + execution_reason = "required external metrics persistence failed" + + # Re-check cancellation after metrics persistence. The heartbeat remains + # active through this point and through close delivery, so a late signal or + # lease loss cannot turn a cancelled execution into a successful close. refused = _cancelled_result(run.id, GLAS_APPROACH, hb) if refused is not None: + if tx_evidence is not None: + refused.detail["repository_transaction"] = tx_evidence + if execution_error == "ExternalMetricsError": + refused.reason = ( + f"{refused.reason}; required external metrics persistence failed" + ) + if hb.monitor.lost or tx is None or tx_evidence is None: + return refused + payload = { + "ok": False, + "approach": GLAS_APPROACH, + "reason": refused.reason, + "repository_transaction": tx_evidence, + } + try: + delivered, state, close_reason = _durable_close( + client, + outbox, + CloseRequest( + run_id=run.id, + transaction_id=tx.transaction_id, + worker_id=client.config.worker_id, + action="fail", + result=payload, + error=refused.reason, + reopen=False, + ), + ) + except (OSError, RuntimeError) as exc: + refused.reason = f"close evidence failure ({type(exc).__name__})" + return refused + if delivered: + refused.ops_state = state + else: + refused.reason = close_reason return refused - if execution_error is not None: - reason = execution_reason or f"profiled execution failed ({execution_error})" + ok = execution_error is None and execution_ok + reason = execution_reason or ( + "profiled execution failed" + if not ok + else "" + ) + payload: dict[str, Any] = { + "ok": ok, + "approach": GLAS_APPROACH, + } + if safe_evidence: + payload["execution_evidence"] = safe_evidence + if reason: + payload["reason"] = reason + if tx_evidence is not None: + payload["repository_transaction"] = tx_evidence + + if tx is None or tx_evidence is None: try: out = client.fail( run.id, error=reason, reopen=execution_reopen, - result={"ok": False, "approach": GLAS_APPROACH, "reason": reason}, + result=payload, ) except OpsRunError as close_exc: return ProcessResult( @@ -403,6 +686,7 @@ def _process_profiled_run( approach=GLAS_APPROACH, ok=False, reason=f"close ops_run failed: {close_exc}; {reason}", + detail=payload, ) return ProcessResult( claimed=True, @@ -411,34 +695,43 @@ def _process_profiled_run( ok=False, reason=reason, ops_state=out.state, + detail=payload, ) - assert gateway_result is not None - evidence = gateway_result["evidence"] - ok = gateway_result["ok"] - reason = str(evidence.get("error") or evidence.get("outcome") or "Glas execution failed") - detail = {"execution_evidence": evidence} - if tx_evidence is not None: - detail["repository_transaction"] = tx_evidence - gateway_result = {**gateway_result, "repository_transaction": tx_evidence} + action = "complete" if ok else "fail" try: - if ok: - out = client.complete(run.id, result=gateway_result) - else: - out = client.fail( - run.id, - error=reason, - reopen=False, - result=gateway_result, - ) - except OpsRunError as exc: + close_request = CloseRequest( + run_id=run.id, + transaction_id=tx.transaction_id, + worker_id=client.config.worker_id, + action=action, + result=payload, + error=reason if action == "fail" else "", + reopen=False, + ) + delivered, state, close_reason = _durable_close( + client, + outbox, + close_request, + ) + except (OSError, RuntimeError) as exc: return ProcessResult( claimed=True, run_id=run.id, approach=GLAS_APPROACH, ok=False, - reason=f"close ops_run failed: {exc}; gateway_ok={ok} {reason}", - detail=detail, + reason=f"close evidence failure ({type(exc).__name__})", + detail=payload, + ) + + if not delivered: + return ProcessResult( + claimed=True, + run_id=run.id, + approach=GLAS_APPROACH, + ok=False, + reason=close_reason, + detail=payload, ) return ProcessResult( @@ -447,8 +740,8 @@ def _process_profiled_run( approach=GLAS_APPROACH, ok=ok, reason="" if ok else reason, - ops_state=out.state, - detail=detail, + ops_state=state, + detail=payload, ) diff --git a/rein_aharness/cli.py b/rein_aharness/cli.py index 186ff85..f6f2011 100644 --- a/rein_aharness/cli.py +++ b/rein_aharness/cli.py @@ -161,6 +161,35 @@ def _cmd_claim_loop(args: argparse.Namespace) -> int: ) +def _cmd_close_outbox(args: argparse.Namespace) -> int: + from rein_aharness.claim_loop import replay_close_outbox + from rein_aharness.close_outbox import CloseOutbox + from rein_aharness.ops_run_client import ActivityCoreOpsClient + + outbox = CloseOutbox() + if args.action == "replay": + if not 1 <= args.limit <= 1000: + print("error: --limit must be between 1 and 1000", file=sys.stderr) + return 2 + report = replay_close_outbox( + ActivityCoreOpsClient(), + outbox, + limit=args.limit, + ) + payload = { + "attempted": report.attempted, + "delivered": report.delivered, + "failed": report.failed, + "quarantined_this_run": report.quarantined, + "remaining": report.remaining, + **outbox.status(), + } + else: + payload = outbox.status() + print(json.dumps(payload, indent=2, sort_keys=True)) + return 1 if payload["pending"] or payload["quarantined"] else 0 + + def _cmd_run(args: argparse.Namespace) -> int: from rein_aharness.intake import IntakeError, IssueCoreClient, poll_next @@ -524,6 +553,18 @@ def main(argv: list[str] | None = None) -> int: claim_loop.add_argument("--no-commit", action="store_true") claim_loop.add_argument("-v", "--verbose", action="store_true") + close_outbox = sub.add_parser( + "close-outbox", + help="Inspect or replay durable Activity Core terminal-close evidence", + ) + close_outbox.add_argument("action", choices=("status", "replay")) + close_outbox.add_argument( + "--limit", + type=int, + default=100, + help="Maximum pending entries to replay (default: 100)", + ) + args = parser.parse_args(argv) if args.command == "validate": @@ -538,6 +579,9 @@ def main(argv: list[str] | None = None) -> int: if args.command == "claim-loop": return _cmd_claim_loop(args) + if args.command == "close-outbox": + return _cmd_close_outbox(args) + if args.command == "run": return _cmd_run(args) diff --git a/rein_aharness/close_outbox.py b/rein_aharness/close_outbox.py index 03754f6..7cc57b7 100644 --- a/rein_aharness/close_outbox.py +++ b/rein_aharness/close_outbox.py @@ -1,8 +1,7 @@ -"""Durable, idempotent close-evidence outbox core. +"""Durable, idempotent Activity Core close-evidence outbox. -The outbox is intentionally not wired into the live claim loop yet. It stores -bounded Activity Core completion/failure intents outside target checkouts so a -future reconciler can retry close delivery without re-running repository work. +It stores bounded completion/failure intents outside target checkouts so the +claim worker can retry close delivery without re-running repository work. """ from __future__ import annotations @@ -32,6 +31,7 @@ _RECORD_KEYS = frozenset( "entry_id", "run_id", "transaction_id", + "worker_id", "action", "result", "error", @@ -61,12 +61,17 @@ class OutboxCorruptError(CloseOutboxError): """A durable record cannot be safely decoded or validated.""" +class PermanentCloseDeliveryError(CloseOutboxError): + """Activity Core conclusively refused a close intent; do not retry it.""" + + @dataclass(frozen=True) class CloseRequest: """One bounded Activity Core completion or failure intent.""" run_id: str transaction_id: str + worker_id: str action: str result: dict[str, Any] error: str = "" @@ -80,6 +85,7 @@ class CloseRequest: ) _validate_identifier(self.run_id, "run_id") _validate_identifier(self.transaction_id, "transaction_id") + _validate_identifier(self.worker_id, "worker_id") if self.action not in {"complete", "fail"}: raise InvalidCloseRequestError("action must be 'complete' or 'fail'") if not isinstance(self.result, dict): @@ -118,6 +124,7 @@ class CloseRequest: "entry_id": self.entry_id, "run_id": self.run_id, "transaction_id": self.transaction_id, + "worker_id": self.worker_id, "action": self.action, "result": self.result, "error": self.error, @@ -128,6 +135,7 @@ class CloseRequest: return CloseRequest( run_id=self.run_id, transaction_id=self.transaction_id, + worker_id=self.worker_id, action=self.action, result=json.loads(json.dumps(self.result)), error=self.error, @@ -203,6 +211,7 @@ class _Record: request = CloseRequest( run_id=value["run_id"], transaction_id=value["transaction_id"], + worker_id=value["worker_id"], action=value["action"], result=value["result"], error=value["error"], @@ -328,6 +337,10 @@ class CloseOutbox: _atomic_write_json(path, trying.payload()) try: deliver(record.request.detached_copy()) + except PermanentCloseDeliveryError as exc: + self._quarantine(path, str(exc)) + quarantined += 1 + continue except Exception as exc: retained = _Record( request=record.request, @@ -370,6 +383,32 @@ class CloseOutbox: with self._locked(): return len(tuple(self.pending_dir.glob("*.json"))) + def status(self) -> dict[str, int]: + """Return bounded operator counts without decoding record contents.""" + with self._locked(): + quarantined = tuple( + path + for path in self.quarantine_dir.glob("*.json") + if not path.name.endswith(".error.json") + ) + return { + "pending": len(tuple(self.pending_dir.glob("*.json"))), + "delivered": len(tuple(self.delivered_dir.glob("*.json"))), + "quarantined": len(quarantined), + } + + def entry_state(self, entry_id: str) -> str | None: + """Return the durable state for one already-validated entry id.""" + _validate_identifier(entry_id, "entry_id") + with self._locked(): + if (self.delivered_dir / f"{entry_id}.json").exists(): + return "delivered" + if (self.pending_dir / f"{entry_id}.json").exists(): + return "pending" + if tuple(self.quarantine_dir.glob(f"{entry_id}.*.json")): + return "quarantined" + return None + def _load(self, path: Path) -> _Record: try: if path.stat().st_size > _MAX_PAYLOAD_BYTES + 8192: diff --git a/rein_aharness/glas_execution.py b/rein_aharness/glas_execution.py index 5bcdbda..b88a6dc 100644 --- a/rein_aharness/glas_execution.py +++ b/rein_aharness/glas_execution.py @@ -8,6 +8,7 @@ with an actionable error if the governed runtime is not installed. from __future__ import annotations +import math from collections.abc import Callable from typing import Any @@ -24,12 +25,86 @@ _SCALAR_REFS = ( "duty_ref", ) _LIST_REFS = ("goal_refs", "resource_envelope_refs") +_EVIDENCE_STRING_FIELDS = ( + "request_id", + "correlation_id", + "actor", + "project", + "target_repo", + "contract_version", + "profile_ref", + "rein_id", + "rein_version", + "model_route", + "resolved_model", + "sandbox_profile", + "sandbox_id", + "tool_profile", + "outcome", + "failure_stage", + "error", + "started_at", + "finished_at", + "commit_sha", + "tool_events_completeness", +) +_EVIDENCE_NUMBER_FIELDS = ( + "duration_s", + "tokens_spent", + "token_budget", + "tool_events_count", +) class GlasExecutionError(RuntimeError): """The authoritative Glas invocation could not produce a GatewayResult.""" +def normalise_execution_evidence_for_close(raw: Any) -> dict[str, Any]: + """Retain only bounded Glas evidence fields safe for durable close state.""" + if not isinstance(raw, dict): + return {} + result: dict[str, Any] = {} + for key in _EVIDENCE_STRING_FIELDS: + value = raw.get(key) + if isinstance(value, str): + result[key] = value[:2000] if key == "error" else value[:500] + for key in _EVIDENCE_NUMBER_FIELDS: + value = raw.get(key) + if ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(value) + and value >= 0 + ): + result[key] = value + artifacts = raw.get("artifacts") + if isinstance(artifacts, list): + result["artifacts"] = [ + item[:1000] + for item in artifacts[:20] + if isinstance(item, str) and item + ] + refs: dict[str, Any] = {} + raw_refs = raw.get("refs") + if isinstance(raw_refs, dict): + for key in _SCALAR_REFS: + value = raw_refs.get(key) + if isinstance(value, str) and value: + refs[key] = value[:500] + for key in _LIST_REFS: + value = raw_refs.get(key) + if isinstance(value, list): + refs[key] = [ + item[:500] + for item in value[:50] + if isinstance(item, str) and item + ] + if refs: + result["refs"] = refs + return result + + def _request_kwargs(run: OpsRun, config: OpsRunConfig, report_to_hub: bool) -> dict[str, Any]: if not run.harness_profile_ref: raise GlasExecutionError(f"ops_run {run.id} has no harness_profile_ref") diff --git a/rein_aharness/ops_run_client.py b/rein_aharness/ops_run_client.py index b0a740c..52ff281 100644 --- a/rein_aharness/ops_run_client.py +++ b/rein_aharness/ops_run_client.py @@ -25,6 +25,7 @@ from typing import Any import httpx from rein_aharness.intake import resolve_target_repo +from rein_aharness.repository_grant import RepositoryGrant, RepositoryGrantError from rein_aharness.taskspec import TaskSpec, TaskSpecError DEFAULT_ACTIVITY_CORE_URL = "http://127.0.0.1:8010" @@ -41,10 +42,12 @@ class OpsRunError(RuntimeError): *, action: str | None = None, status_code: int | None = None, + code: str | None = None, ) -> None: super().__init__(message) self.action = action self.status_code = status_code + self.code = code @property def lease_rejected(self) -> bool: @@ -73,11 +76,25 @@ class OpsRun: approach_hint: str | None = None harness_profile_ref: str | None = None execution_refs: dict[str, Any] = field(default_factory=dict) + repository_grant: RepositoryGrant | None = None result: dict[str, Any] = field(default_factory=dict) + close_disposition: str | None = None + close_intent_digest: str | None = None raw: dict[str, Any] = field(default_factory=dict) @classmethod def from_api(cls, data: dict[str, Any]) -> "OpsRun": + grant = None + if data.get("repository_grant") is not None: + try: + grant = RepositoryGrant.from_mapping(data["repository_grant"]) + except RepositoryGrantError as exc: + run_id = str(data.get("id") or "unknown")[:200] + raise OpsRunError( + f"ops_run {run_id} has an invalid repository_grant", + action="decode", + code="invalid_repository_grant", + ) from exc return cls( id=str(data.get("id") or ""), activity_definition_id=str(data.get("activity_definition_id") or ""), @@ -97,7 +114,14 @@ class OpsRun: approach_hint=data.get("approach_hint"), harness_profile_ref=data.get("harness_profile_ref"), execution_refs=dict(data.get("execution_refs") or {}), + repository_grant=grant, result=dict(data.get("result") or {}), + close_disposition=_bounded_optional_string( + data.get("close_disposition"), 80 + ), + close_intent_digest=_bounded_optional_string( + data.get("close_intent_digest"), 128 + ), raw=data, ) @@ -279,6 +303,7 @@ class ActivityCoreOpsClient: f"{action} ops_run {run_id} failed: HTTP {exc.response.status_code}", action=action, status_code=exc.response.status_code, + code=_response_error_code(exc.response), ) from exc except httpx.HTTPError as exc: raise OpsRunError( @@ -313,6 +338,7 @@ def ops_run_to_taskspec( hub_task_id=run.id, completion_event_type=completion_event_type, timeout_seconds=timeout_seconds, + repository_grant=run.repository_grant, ) @@ -325,3 +351,19 @@ def resolve_ops_target(run: OpsRun, config: OpsRunConfig | None = None) -> Path: repo_map=cfg.repo_map, repo_roots=cfg.repo_roots, ) + + +def _response_error_code(response: httpx.Response) -> str | None: + try: + payload = response.json() + except (ValueError, TypeError): + return None + detail = payload.get("detail") if isinstance(payload, dict) else None + code = detail.get("code") if isinstance(detail, dict) else None + return _bounded_optional_string(code, 80) + + +def _bounded_optional_string(value: Any, limit: int) -> str | None: + if not isinstance(value, str) or not value: + return None + return value[:limit] diff --git a/rein_aharness/repository_transaction.py b/rein_aharness/repository_transaction.py index d11012d..2eb9132 100644 --- a/rein_aharness/repository_transaction.py +++ b/rein_aharness/repository_transaction.py @@ -2,8 +2,7 @@ Wired into `run_task`, profile-absent `execute_approach` mutators, and the profiled claim path under REINAH-WP-0003-T02 / ADR-002. Explicit local -TaskSpec grants activate repository acceptance; queued/profiled grant carriage -remains an upstream contract dependency. +TaskSpec and profiled Activity Core grants activate repository acceptance. """ from __future__ import annotations @@ -318,6 +317,24 @@ class RepositoryTransaction: "branch-changed", f"expected={baseline.branch or 'detached'} actual={post.branch or 'detached'}", ) + if not post.clean: + raise RepositoryAcceptanceError( + "dirty-post-state", + f"entries={post.dirty_entries} status_digest={post.status_digest}", + ) + if post.remote_refs != baseline.remote_refs: + raise RepositoryAcceptanceError( + "remote-refs-changed", + "local remote-tracking refs moved during the transaction", + ) + if ( + post.protected_git_metadata_digest + != baseline.protected_git_metadata_digest + ): + raise RepositoryAcceptanceError( + "git-metadata-changed", + "protected Git config, hooks, or info metadata changed", + ) if post.head == baseline.head: raise RepositoryAcceptanceError("head-unchanged", "no new commit to accept") @@ -378,25 +395,6 @@ class RepositoryTransaction: f"count={len(invalid_paths)} paths={visible[:220]}", ) - if not post.clean: - raise RepositoryAcceptanceError( - "dirty-post-state", - f"entries={post.dirty_entries} status_digest={post.status_digest}", - ) - if post.remote_refs != baseline.remote_refs: - raise RepositoryAcceptanceError( - "remote-refs-changed", - "local remote-tracking refs moved during the transaction", - ) - if ( - post.protected_git_metadata_digest - != baseline.protected_git_metadata_digest - ): - raise RepositoryAcceptanceError( - "git-metadata-changed", - "protected Git config, hooks, or info metadata changed", - ) - acceptance = RepositoryAcceptance( policy_id=policy.policy_id, head=post.head, diff --git a/tests/test_claim_loop.py b/tests/test_claim_loop.py index be85a93..c592fdf 100644 --- a/tests/test_claim_loop.py +++ b/tests/test_claim_loop.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import subprocess import time from pathlib import Path @@ -14,14 +15,17 @@ from rein_aharness.claim_loop import ( process_one, poll_peek, ) +from rein_aharness.close_outbox import CloseOutbox, CloseRequest from rein_aharness.execution_cancel import ExecutionCancel, active_cancel from rein_aharness.glas_execution import GLAS_APPROACH, GlasExecutionError +from rein_aharness.metrics import external_metrics_dir from rein_aharness.ops_run_client import ( ActivityCoreOpsClient, OpsRun, OpsRunConfig, OpsRunError, ) +from rein_aharness.repository_grant import RepositoryGrant from rein_aharness.repository_transaction import RepositoryTransaction @@ -75,6 +79,45 @@ def _profiled_case( return repo, run, client +def _grant(*paths: str) -> RepositoryGrant: + return RepositoryGrant.from_mapping( + { + "version": "1", + "allowed_paths": list(paths or ("docs/",)), + "commit_count": {"min": 1, "max": 1}, + "publish": False, + } + ) + + +def _commit_file(repo: Path, relative: str, value: str = "result\n") -> str: + path = repo / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(value, encoding="utf-8") + subprocess.run(["git", "add", relative], cwd=repo, check=True) + subprocess.run( + [ + "git", + "-c", + "user.email=test@example.invalid", + "-c", + "user.name=test", + "commit", + "-qm", + "profiled result", + ], + cwd=repo, + check=True, + ) + return subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=repo, + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + def test_process_one_empty() -> None: client = MagicMock(spec=ActivityCoreOpsClient) client.config = OpsRunConfig(worker_id="w", lease_seconds=90) @@ -150,6 +193,31 @@ def test_process_one_failure_reopens() -> None: assert client.fail.call_args.kwargs["reopen"] is True +def test_grant_without_authoritative_profile_refuses_before_execution() -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + run = _claimed_run() + run.repository_grant = _grant("docs/") + client.claim.return_value = [run] + client.fail.return_value = OpsRun( + id=run.id, + activity_definition_id="def", + idempotency_key="k", + target_repo=run.target_repo, + title=run.title, + description="", + state="failed", + ) + + with patch("rein_aharness.claim_loop.execute_approach") as execute: + result = process_one(client) + + assert result.ok is False + assert "requires an authoritative harness_profile_ref" in result.reason + execute.assert_not_called() + assert client.fail.call_args.kwargs["reopen"] is False + + def test_process_one_refuses_close_after_lease_loss() -> None: client = MagicMock(spec=ActivityCoreOpsClient) client.config = OpsRunConfig(worker_id="w", lease_seconds=90) @@ -340,7 +408,8 @@ def test_profiled_run_uses_glas_and_completes_with_full_result( complete_run_id = client.complete.call_args.args[0] complete_result = client.complete.call_args.kwargs["result"] assert complete_run_id == run.id - assert complete_result["evidence"] == gateway_result["evidence"] + assert complete_result["execution_evidence"] == gateway_result["evidence"] + assert "tool_output" not in complete_result assert complete_result["repository_transaction"] == transaction client.fail.assert_not_called() select.assert_not_called() @@ -379,7 +448,7 @@ def test_profile_refusal_fails_terminally_without_legacy_fallback( execute.assert_not_called() -def test_profiled_signal_cancellation_releases_repository_lock( +def test_profiled_signal_cancellation_releases_lock_and_durably_fails( tmp_path: Path, ) -> None: repo, _run, client = _profiled_case(tmp_path) @@ -397,7 +466,9 @@ def test_profiled_signal_cancellation_releases_repository_lock( assert result.ok is False assert result.reason == "execution cancelled (signal)" client.complete.assert_not_called() - client.fail.assert_not_called() + client.fail.assert_called_once() + assert client.fail.call_args.kwargs["reopen"] is False + assert "repository_transaction" in client.fail.call_args.kwargs["result"] with RepositoryTransaction(repo) as retry: assert retry.locked is True @@ -416,12 +487,240 @@ def test_profiled_close_failure_happens_after_repository_lock_release( result = process_one(client) assert result.ok is False - assert result.reason.startswith("close ops_run failed:") + assert result.reason == "close evidence remains pending" assert "repository_transaction" in result.detail with RepositoryTransaction(repo) as retry: assert retry.locked is True +def test_granted_profile_accepts_commit_records_metrics_and_durable_close( + tmp_path: Path, +) -> None: + repo, run, client = _profiled_case(tmp_path) + run.repository_grant = _grant("docs/") + client.complete.return_value = OpsRun( + id=run.id, + activity_definition_id="def", + idempotency_key="k", + target_repo=run.target_repo, + title=run.title, + description="", + state="succeeded", + close_disposition="applied", + ) + outbox = CloseOutbox(state_dir=tmp_path / "state") + + def execute(*_args, **_kwargs): + head = _commit_file(repo, "docs/result.md") + return { + "ok": True, + "evidence": { + "outcome": "succeeded", + "commit_sha": head, + "provider_response": {"secret": "must-not-persist"}, + }, + "tool_output": "must-not-persist", + } + + with patch("rein_aharness.claim_loop.execute_profiled_run", side_effect=execute): + result = process_one(client, outbox=outbox) + + assert result.ok is True + transaction = result.detail["repository_transaction"] + assert transaction["acceptance"]["accepted"] is True + assert transaction["acceptance"]["changed_paths"] == ["docs/result.md"] + assert transaction["repository_grant"]["grant_id"] == run.repository_grant.grant_id + assert transaction["metrics"]["session_id"] == transaction["transaction_id"] + assert outbox.status() == {"pending": 0, "delivered": 1, "quarantined": 0} + + close_payload = client.complete.call_args.kwargs["result"] + serialized = json.dumps(close_payload) + assert "tool_output" not in serialized + assert "provider_response" not in serialized + assert close_payload["repository_transaction"]["acceptance"]["accepted"] is True + + ledger = external_metrics_dir(repo, "rein-aharness") / "executions.jsonl" + record = json.loads(ledger.read_text(encoding="utf-8").strip()) + assert record["success"] is True + assert record["session_id"] == transaction["transaction_id"] + + +def test_granted_profile_rejects_out_of_grant_commit_and_closes_failed( + tmp_path: Path, +) -> None: + repo, run, client = _profiled_case(tmp_path) + run.repository_grant = _grant("docs/") + client.fail.return_value = OpsRun( + id=run.id, + activity_definition_id="def", + idempotency_key="k", + target_repo=run.target_repo, + title=run.title, + description="", + state="failed", + close_disposition="applied", + ) + + def execute(*_args, **_kwargs): + head = _commit_file(repo, "UNRELATED.md") + return {"ok": True, "evidence": {"outcome": "succeeded", "commit_sha": head}} + + with patch("rein_aharness.claim_loop.execute_profiled_run", side_effect=execute): + result = process_one(client, outbox=CloseOutbox(state_dir=tmp_path / "state")) + + assert result.ok is False + assert result.reason.startswith("repository acceptance failed: path-not-granted") + client.complete.assert_not_called() + client.fail.assert_called_once() + close_payload = client.fail.call_args.kwargs["result"] + assert close_payload["repository_transaction"]["repository_grant"]["grant_id"] + assert "acceptance" not in close_payload["repository_transaction"] + assert client.fail.call_args.kwargs["reopen"] is False + + +def test_granted_profile_metrics_failure_prevents_successful_close( + tmp_path: Path, + monkeypatch, +) -> None: + repo, run, client = _profiled_case(tmp_path) + run.repository_grant = _grant("docs/") + client.fail.return_value = OpsRun( + id=run.id, + activity_definition_id="def", + idempotency_key="k", + target_repo=run.target_repo, + title=run.title, + description="", + state="failed", + ) + + def execute(*_args, **_kwargs): + head = _commit_file(repo, "docs/result.md") + return {"ok": True, "evidence": {"outcome": "succeeded", "commit_sha": head}} + + def fail_metrics(*_args, **_kwargs): + raise OSError("state volume unavailable") + + monkeypatch.setattr( + "rein_aharness.claim_loop.metrics.record_external_execution", + fail_metrics, + ) + with patch("rein_aharness.claim_loop.execute_profiled_run", side_effect=execute): + result = process_one(client, outbox=CloseOutbox(state_dir=tmp_path / "state")) + + assert result.ok is False + assert result.reason == "required external metrics persistence failed" + client.complete.assert_not_called() + close_payload = client.fail.call_args.kwargs["result"] + assert close_payload["repository_transaction"]["acceptance"]["accepted"] is True + assert "metrics" not in close_payload["repository_transaction"] + + +def test_response_lost_close_replays_without_reexecuting_workload( + tmp_path: Path, +) -> None: + repo, run, client = _profiled_case(tmp_path) + run.repository_grant = _grant("docs/") + outbox = CloseOutbox(state_dir=tmp_path / "state") + reconciled = OpsRun( + id=run.id, + activity_definition_id="def", + idempotency_key="k", + target_repo=run.target_repo, + title=run.title, + description="", + state="succeeded", + close_disposition="reconciled", + ) + client.complete.side_effect = [ + OpsRunError("complete transport failed", action="complete"), + reconciled, + ] + gateway_calls = 0 + + def execute(*_args, **_kwargs): + nonlocal gateway_calls + gateway_calls += 1 + head = _commit_file(repo, "docs/result.md") + return { + "ok": True, + "evidence": {"outcome": "succeeded", "commit_sha": head}, + } + + with patch("rein_aharness.claim_loop.execute_profiled_run", side_effect=execute): + first = process_one(client, outbox=outbox) + + assert first.reason == "close evidence remains pending" + assert outbox.status()["pending"] == 1 + client.claim.return_value = [] + + second = process_one(client, outbox=outbox) + + assert second.empty is True + assert gateway_calls == 1 + assert client.complete.call_count == 2 + assert outbox.status() == {"pending": 0, "delivered": 1, "quarantined": 0} + commit_count = subprocess.run( + ["git", "rev-list", "--count", "HEAD"], + cwd=repo, + check=True, + capture_output=True, + text=True, + ).stdout.strip() + assert commit_count == "2" + + +def test_pending_close_failure_blocks_new_claim(tmp_path: Path) -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + client.complete.side_effect = OpsRunError("transport failed", action="complete") + outbox = CloseOutbox(state_dir=tmp_path / "state") + outbox.enqueue( + CloseRequest( + run_id="run-1", + transaction_id="tx-1", + worker_id="w", + action="complete", + result={"ok": True}, + ) + ) + + result = process_one(client, outbox=outbox) + + assert result.claimed is False + assert result.retry_full_interval is True + assert "remains pending" in result.reason + client.claim.assert_not_called() + + +def test_permanent_close_conflict_is_quarantined_before_claim(tmp_path: Path) -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + client.complete.side_effect = OpsRunError( + "conflict", + action="complete", + status_code=409, + code="terminal_conflict", + ) + client.claim.return_value = [] + outbox = CloseOutbox(state_dir=tmp_path / "state") + outbox.enqueue( + CloseRequest( + run_id="run-1", + transaction_id="tx-1", + worker_id="w", + action="complete", + result={"ok": True}, + ) + ) + + result = process_one(client, outbox=outbox) + + assert result.empty is True + assert outbox.status() == {"pending": 0, "delivered": 0, "quarantined": 1} + client.claim.assert_called_once() + + def test_poll_peek_reports_authoritative_profile_route() -> None: client = MagicMock(spec=ActivityCoreOpsClient) run = _claimed_run() diff --git a/tests/test_close_outbox.py b/tests/test_close_outbox.py index ffabc8b..ae6602b 100644 --- a/tests/test_close_outbox.py +++ b/tests/test_close_outbox.py @@ -16,6 +16,7 @@ from rein_aharness.close_outbox import ( InvalidCloseRequestError, OutboxConflictError, OutboxCorruptError, + PermanentCloseDeliveryError, ) @@ -23,6 +24,7 @@ def _request( *, run_id: str = "run-1", transaction_id: str = "tx-1", + worker_id: str = "worker-1", action: str = "complete", result: dict[str, object] | None = None, error: str = "", @@ -31,6 +33,7 @@ def _request( return CloseRequest( run_id=run_id, transaction_id=transaction_id, + worker_id=worker_id, action=action, result=result or {"ok": True, "accepted_commit": "a" * 40}, error=error, @@ -57,6 +60,7 @@ def test_enqueue_uses_private_external_atomic_record(tmp_path: Path) -> None: assert payload["state"] == "pending" assert payload["attempts"] == 0 assert payload["entry_id"] == receipt.entry_id + assert payload["worker_id"] == "worker-1" assert not tuple(outbox.pending_dir.glob("*.tmp")) @@ -164,6 +168,21 @@ def test_process_interrupt_leaves_attempt_durable_and_pending(tmp_path: Path) -> assert payload["last_error"] is None +def test_permanent_delivery_refusal_is_quarantined(tmp_path: Path) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + receipt = outbox.enqueue(_request()) + + def conflict(_: CloseRequest) -> None: + raise PermanentCloseDeliveryError("terminal_conflict") + + report = outbox.replay(conflict) + + assert report.quarantined == 1 + assert report.remaining == 0 + assert outbox.entry_state(receipt.entry_id) == "quarantined" + assert outbox.status() == {"pending": 0, "delivered": 0, "quarantined": 1} + + def test_corrupt_pending_record_is_preserved_in_quarantine(tmp_path: Path) -> None: outbox = CloseOutbox(state_dir=tmp_path / "state") request = _request() @@ -232,6 +251,7 @@ outbox = CloseOutbox(state_dir=Path(sys.argv[1])) receipt = outbox.enqueue(CloseRequest( run_id="run-shared", transaction_id="tx-shared", + worker_id="worker-1", action="complete", result={"ok": True}, )) @@ -279,6 +299,7 @@ def test_replay_limit_is_bounded_and_leaves_remaining_entries(tmp_path: Path) -> [ ({"run_id": "bad/id"}, "run_id"), ({"transaction_id": "with space"}, "transaction_id"), + ({"worker_id": "with space"}, "worker_id"), ({"action": "cancel"}, "action"), ({"action": "complete", "error": "not allowed"}, "cannot carry"), ({"action": "complete", "reopen": True}, "cannot carry"), @@ -292,6 +313,7 @@ def test_close_request_rejects_ambiguous_identity_or_action( values: dict[str, object] = { "run_id": "run-1", "transaction_id": "tx-1", + "worker_id": "worker-1", "action": "complete", "result": {"ok": True}, "error": "", @@ -325,6 +347,7 @@ def test_close_request_rejects_unbounded_or_non_json_result( CloseRequest( run_id="run-1", transaction_id="tx-1", + worker_id="worker-1", action="complete", result=result, # type: ignore[arg-type] ) diff --git a/tests/test_close_outbox_cli.py b/tests/test_close_outbox_cli.py new file mode 100644 index 0000000..8b845e7 --- /dev/null +++ b/tests/test_close_outbox_cli.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from rein_aharness.cli import main +from rein_aharness.close_outbox import CloseOutbox, CloseRequest + + +def test_close_outbox_status_reports_pending_and_sets_failure_exit( + tmp_path: Path, + capsys, +) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "runtime-state") + outbox.enqueue( + CloseRequest( + run_id="run-1", + transaction_id="tx-1", + worker_id="worker-1", + action="complete", + result={"ok": True}, + ) + ) + + assert main(["close-outbox", "status"]) == 1 + payload = json.loads(capsys.readouterr().out) + assert payload == {"delivered": 0, "pending": 1, "quarantined": 0} + + +def test_close_outbox_replay_delivers_pending_entry( + tmp_path: Path, + capsys, + monkeypatch, +) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "runtime-state") + outbox.enqueue( + CloseRequest( + run_id="run-1", + transaction_id="tx-1", + worker_id="worker-1", + action="complete", + result={"ok": True}, + ) + ) + + class Client: + class Config: + worker_id = "worker-1" + + config = Config() + + def complete(self, run_id: str, *, result: dict): + assert run_id == "run-1" + assert result == {"ok": True} + return None + + monkeypatch.setattr( + "rein_aharness.ops_run_client.ActivityCoreOpsClient", + Client, + ) + + assert main(["close-outbox", "replay"]) == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["delivered"] == 1 + assert payload["pending"] == 0 + + +def test_close_outbox_replay_rejects_invalid_limit(capsys) -> None: + assert main(["close-outbox", "replay", "--limit", "0"]) == 2 + assert "between 1 and 1000" in capsys.readouterr().err diff --git a/tests/test_glas_execution.py b/tests/test_glas_execution.py index 32c2616..f05d92a 100644 --- a/tests/test_glas_execution.py +++ b/tests/test_glas_execution.py @@ -7,7 +7,12 @@ from pathlib import Path import pytest -from rein_aharness.glas_execution import GLAS_ACTOR, GlasExecutionError, execute_profiled_run +from rein_aharness.glas_execution import ( + GLAS_ACTOR, + GlasExecutionError, + execute_profiled_run, + normalise_execution_evidence_for_close, +) from rein_aharness.ops_run_client import OpsRun, OpsRunConfig @@ -93,6 +98,29 @@ def test_profiled_run_rejects_invalid_gateway_result(tmp_path: Path) -> None: ) +def test_close_evidence_normalizer_drops_direct_and_unknown_values() -> None: + result = normalise_execution_evidence_for_close( + { + "outcome": "succeeded", + "profile_ref": "harness.agent-dev-local@1.0.0", + "artifacts": ["docs/result.md", {"secret": "drop"}], + "refs": { + "assignment_ref": "assignment:1", + "goal_refs": ["goal:1", {"secret": "drop"}], + "api_key": "drop", + }, + "provider_response": {"secret": "drop"}, + } + ) + + assert result == { + "profile_ref": "harness.agent-dev-local@1.0.0", + "outcome": "succeeded", + "artifacts": ["docs/result.md"], + "refs": {"assignment_ref": "assignment:1", "goal_refs": ["goal:1"]}, + } + + def test_profiled_actor_validates_against_real_glas_and_sandboxer(tmp_path: Path) -> None: contract = pytest.importorskip("glas_harness.contract") sandbox_models = pytest.importorskip("sandboxer.models") diff --git a/tests/test_ops_run_client.py b/tests/test_ops_run_client.py index 9ef4397..9437a59 100644 --- a/tests/test_ops_run_client.py +++ b/tests/test_ops_run_client.py @@ -36,6 +36,12 @@ def test_ops_run_from_api() -> None: "correlation_id": "corr-1", "goal_refs": ["goal:42@1"], }, + "repository_grant": { + "version": "1", + "allowed_paths": ["docs/", "README.md"], + "commit_count": {"min": 1, "max": 1}, + "publish": False, + }, } ) assert row.target_repo == "freedom-intelligence" @@ -45,6 +51,25 @@ def test_ops_run_from_api() -> None: "correlation_id": "corr-1", "goal_refs": ["goal:42@1"], } + assert row.repository_grant is not None + assert row.repository_grant.grant_id == "af2e7c8275c9ba8c8f78485067f9608e" + + +def test_ops_run_from_api_refuses_invalid_repository_grant() -> None: + with pytest.raises(OpsRunError, match="invalid repository_grant") as exc_info: + OpsRun.from_api( + { + "id": "run-1", + "repository_grant": { + "version": "1", + "allowed_paths": ["../escape"], + "commit_count": {"min": 1, "max": 1}, + "publish": False, + }, + } + ) + + assert exc_info.value.code == "invalid_repository_grant" def test_config_from_env(monkeypatch: pytest.MonkeyPatch) -> None: @@ -126,6 +151,38 @@ def test_complete_and_fail() -> None: assert post.call_args.kwargs["json"]["reopen"] is True +def test_complete_parses_reconciliation_and_machine_error_code() -> None: + cfg = OpsRunConfig(base_url="http://example.test", worker_id="w") + client = ActivityCoreOpsClient(cfg) + accepted = MagicMock() + accepted.raise_for_status = MagicMock() + accepted.json.return_value = { + "id": "r1", + "activity_definition_id": "d", + "idempotency_key": "k", + "title": "t", + "description": "", + "state": "succeeded", + "labels": [], + "close_disposition": "reconciled", + } + with patch("rein_aharness.ops_run_client.httpx.post", return_value=accepted): + out = client.complete("r1", result={"ok": True}) + assert out.close_disposition == "reconciled" + + request = httpx.Request("POST", "http://example.test/ops-runs/r1/complete") + conflict = httpx.Response( + 409, + request=request, + json={"detail": {"code": "terminal_conflict", "message": "conflict"}}, + ) + with patch("rein_aharness.ops_run_client.httpx.post", return_value=conflict): + with pytest.raises(OpsRunError) as exc_info: + client.complete("r1", result={"ok": True}) + assert exc_info.value.status_code == 409 + assert exc_info.value.code == "terminal_conflict" + + def test_claim_http_error() -> None: cfg = OpsRunConfig(base_url="http://example.test", worker_id="w") client = ActivityCoreOpsClient(cfg) @@ -175,6 +232,30 @@ def test_ops_run_to_taskspec(tmp_path: Path) -> None: assert spec.hub_task_id == "r1" +def test_ops_run_to_taskspec_carries_repository_grant(tmp_path: Path) -> None: + import subprocess + + repo = tmp_path / "controlled" + repo.mkdir() + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + run = OpsRun.from_api( + { + "id": "r1", + "target_repo": "controlled", + "repository_grant": { + "version": "1", + "allowed_paths": ["docs/"], + "commit_count": {"min": 1, "max": 1}, + "publish": False, + }, + } + ) + + spec = ops_run_to_taskspec(run, OpsRunConfig(repo_roots=(str(tmp_path),))) + + assert spec.repository_grant is run.repository_grant + + def test_ops_run_to_taskspec_missing_repo() -> None: run = OpsRun( id="r1", diff --git a/tests/test_repository_transaction.py b/tests/test_repository_transaction.py index 81c579a..ccd7d4c 100644 --- a/tests/test_repository_transaction.py +++ b/tests/test_repository_transaction.py @@ -285,6 +285,21 @@ def test_acceptance_rejects_dirty_post_state(tmp_path: Path) -> None: assert excinfo.value.code == "dirty-post-state" +def test_acceptance_checks_dirty_post_state_before_unchanged_head( + tmp_path: Path, +) -> None: + repo = _make_repo(tmp_path) + + with RepositoryTransaction(repo, state_dir=tmp_path / "state") as tx: + (repo / "uncommitted.txt").write_text("left dirty\n", encoding="utf-8") + with pytest.raises(RepositoryAcceptanceError) as excinfo: + tx.validate_acceptance( + RepositoryAcceptancePolicy(allowed_paths=("docs/",)) + ) + + assert excinfo.value.code == "dirty-post-state" + + def test_acceptance_rejects_remote_tracking_ref_movement(tmp_path: Path) -> None: repo = _make_repo(tmp_path) subprocess.run( diff --git a/workplans/REINAH-WP-0003-governed-runtime-integrity.md b/workplans/REINAH-WP-0003-governed-runtime-integrity.md index 4dbc4de..b7a069b 100644 --- a/workplans/REINAH-WP-0003-governed-runtime-integrity.md +++ b/workplans/REINAH-WP-0003-governed-runtime-integrity.md @@ -278,7 +278,7 @@ evidence, and lease-cancellation cases, all T02 exit criteria are met. T02 is ```task id: REINAH-WP-0003-T03 -status: wait +status: done priority: high state_hub_task_id: "6297dee4-c10f-5d5f-8d7c-7edfafa27e7f" ``` @@ -417,6 +417,40 @@ terminal reconciliation in State Hub message `6c4189a5-5971-4eb9-a7bd-62fe368f9a02`. Live outbox activation stays fail-closed until that exact upstream contract is available. +### Activity Core contract consumption — 2026-09-04 + +Reviewed Activity Core implementation `b63131e`, subsequently published on +its `origin/main` through `52f2ac9`. Independent verification passes all 73 +focused grant/evidence/close tests and the full Activity Core suite (`493 +passed, 1 skipped`). It satisfies the requested upstream contract: migration +`0010` adds separately typed grant and terminal intent fields; definition and +emission boundaries validate v1 authority; results retain bounded transaction, +grant, acceptance, and metrics identity; and exact same-owner terminal repeats +return `reconciled` while conflicts remain machine-distinct. + +The rein claim client now parses the queued grant strictly and carries it +without deriving authority. A grant-bearing row requires an authoritative +profile. Under the existing lease-bound repository transaction, profiled runs +validate every successful commit against that grant, persist transaction-linked +external metrics, and submit only bounded execution/transaction evidence. +Unknown provider fields and direct tool output never enter durable close state. + +Activated the close outbox for profiled terminal results. The exact run, +transaction, worker, action, reopen decision, and bounded result are persisted +before the API call. Pending close-only work replays before another claim, an +Activity Core `applied` or `reconciled` response marks it delivered, transient +or unclassified failures remain pending, and conclusive owner/lease/state/ +evidence/terminal conflicts move to quarantine. `rein-aharness close-outbox +status|replay` supplies bounded operator controls. Tests simulate a lost close +response and prove the same durable intent reconciles without a second gateway +or repository execution. + +Final contract review and verification pass: 132 focused tests cover the +repository, grant, queue, evidence, outbox, CLI, and claim paths; the full rein +suite passes (`241 passed, 1 skipped`). T03 is `done`. Deployment of Activity +Core migration `0010` and the compatible worker artifact remains tracked under +T05/T06 rather than being represented as completed here. + ## Remove tenant logic from the shared runtime ```task