diff --git a/.custodian-brief.md b/.custodian-brief.md index 8133138..b50498f 100644 --- a/.custodian-brief.md +++ b/.custodian-brief.md @@ -2,21 +2,20 @@ # Custodian Brief — rein-aharness **Domain:** agents -**Last synced:** 2026-08-25 20:42 UTC +**Last synced:** 2026-09-04 09:00 UTC **State Hub:** http://127.0.0.1:8000 *(adjust if running on a remote machine)* ## Active Workstreams ### Governed runtime integrity and intent convergence -Progress: 0/6 done | workplan_id: `eba2eff1-10a7-50a3-a70b-14e7d398f27f` +Progress: 1/6 done | workplan_id: `eba2eff1-10a7-50a3-a70b-14e7d398f27f` **Open tasks:** - ! Verify accepted commits and reconcile metrics/reporting `6297dee4` -- ! Remove tenant logic from the shared runtime `f6c7f80c` -- ! Align deployment, recovery, and conformance with the supported runtime `c4a3f08f` - ! Re-prove one governed profiled run and close residuals `68a13c4f` -- ► Establish the governance and enforcement baseline `596fc138` - ► Make each repository run a lease-bound transaction `31da4226` +- · Remove tenant logic from the shared runtime `f6c7f80c` +- · Align deployment, recovery, and conformance with the supported runtime `c4a3f08f` --- ## MCP Orientation (when available) diff --git a/INTENT.md b/INTENT.md index 51cca1d..5b361f9 100644 --- a/INTENT.md +++ b/INTENT.md @@ -46,10 +46,15 @@ place.** verifies the local commit, reports to the State Hub and to the instance's kaizen metrics. -The harness is the **only credential holder and the only policy -enforcement point** for agent sessions. Instances declare policy; the -harness enforces it. Instances pin a harness major version; upgrades -happen centrally. +Among tenant repositories, blueprints, and scheduling records, only the +selected rein may receive session credential material, and it holds that +material only for the bounded session (ADR-002). The harness is the +**repository-transaction and rein-local tool-policy enforcement point** +inside a chained grant — not the only credential holder or the only +policy enforcement point in the estate. Queue, profile, sandbox, +credential custody/authorization, and provider boundaries enforce their +own grants; this runtime must not weaken them. Instances declare policy +and pin a harness major version; upgrades happen centrally. ## Strategic role diff --git a/SCOPE.md b/SCOPE.md index 6da3b97..268c0a3 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -85,8 +85,10 @@ retains a legacy set of local task executors for unattended repository work. executable bwrap reachability, mounted rein runtime, and model egress remain with upstream residual `GLAS-IN-0002`. - Direct legacy agent sessions rely on Claude Code's own tool mediation and - the host checkout. They are not an OS-level sandbox, do not lock the repo, - and do not prove that no push/network activity occurred after the run. + 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); commit acceptance + is still the `HEAD`-changed check until T03. - Commit acceptance currently means only that `HEAD` changed. It does not yet validate the changed-file set, branch/parent shape, clean working tree, remote state, or provenance of the new commit. Metrics are written after diff --git a/docs/adr/ADR-001-agent-harness-architecture.md b/docs/adr/ADR-001-agent-harness-architecture.md index cf4de79..1e62c84 100644 --- a/docs/adr/ADR-001-agent-harness-architecture.md +++ b/docs/adr/ADR-001-agent-harness-architecture.md @@ -35,8 +35,11 @@ control's operating rhythm depended on a workstation cron bridge credentials, no tool enumerations. 3. **Blueprints stay in kaizen-agentic; scheduling stays in activity-core; provider abstraction stays in llm-connect.** The - harness consumes all three and is the only credential holder and - policy enforcement point for agent sessions. + harness consumes all three. Per ADR-002 it is the + repository-transaction and rein-local tool-policy enforcement point + within a chained grant, not the only credential holder or the only + policy enforcement point in the estate. Tenant repos, blueprints, and + schedules still must not hold session credentials or tool wiring. 4. The 2026-07-17 executor-worker prototype is adopted as the harness seed (`agent_harness/` package). binky-control is tenant #1. @@ -55,3 +58,6 @@ Details: [docs/architecture.md](../architecture.md). - Repo-boundary discipline continues: any pressure to add scheduling, blueprint authoring, or tenant-specific logic here is a signal to extend the neighbor repo or the instance model instead. +- ADR-002 (accepted 2026-08-29) amends the credential-holder and + policy-enforcement slogans in decision 3. The instance/runtime split + and the prohibition on tenant-held credentials are unchanged. diff --git a/docs/adr/ADR-002-governed-execution-responsibility-chain.md b/docs/adr/ADR-002-governed-execution-responsibility-chain.md index 239f1b8..34fa485 100644 --- a/docs/adr/ADR-002-governed-execution-responsibility-chain.md +++ b/docs/adr/ADR-002-governed-execution-responsibility-chain.md @@ -1,19 +1,28 @@ --- id: ADR-002 title: Governed execution is a responsibility chain, not a single enforcement point -status: proposed +status: accepted date: "2026-08-23" -workplan_task: HARNESS-WP-0003-T01 +accepted: "2026-08-29" +workplan_task: REINAH-WP-0003-T01 --- # ADR-002 — Governed execution responsibility chain ## Status -Proposed. Acceptance requires acknowledgement from the owners of Activity Core, -Glas, sand-boxer, and llm-connect. Until then this document records the -implementation target for `HARNESS-WP-0003` but does not amend `INTENT.md` or -supersede accepted ADR-001. +Accepted 2026-08-29 by operator Bernd Worsch. This is the implementation +contract for `REINAH-WP-0003`. It amends the slogan wording in `INTENT.md` and +ADR-001 on the same date; it does not replace ADR-001's instance/runtime split +or the prohibition on tenant-held credentials and tenant-specific runtime +wiring. + +The proposed text required sibling-owner acknowledgement before acceptance. +Glas had already approved the boundary; Activity Core had deployed the queue +and lease behavior it asked to see in writing. The operator accepted the +substance without waiting for remaining exact-revision acknowledgements from +Activity Core, sand-boxer, and llm-connect. Those recordings remain welcome; +they are no longer a gate. Sibling objection still changes the matrix. ## Context @@ -211,4 +220,8 @@ Reviews requiring follow-up: live spoof-refusal evidence, and no manufactured late-close run. Those facts are incorporated in this revision for final Activity Core acknowledgement. -Pending: revised Activity Core acknowledgement, sand-boxer, and llm-connect. +Pending as recording only, not a gate: revised Activity Core +acknowledgement of this accepted text, sand-boxer, and llm-connect. + +Operator acceptance: Bernd Worsch, 2026-08-29, session decision to accept +ADR-002 as written and proceed with T02 wiring. diff --git a/rein_aharness/adapter.py b/rein_aharness/adapter.py index ba4bd99..6e0efe6 100644 --- a/rein_aharness/adapter.py +++ b/rein_aharness/adapter.py @@ -33,6 +33,11 @@ from llm_connect.claude_code import ClaudeCodeAdapter from llm_connect.exceptions import LLMSubprocessError, LLMTimeoutError from llm_connect.models import LLMResponse, RunConfig +from rein_aharness.execution_cancel import ( + ExecutionCancel, + ExecutionCancelled, + resolve_cancel, +) from rein_aharness.profiles import ToolProfile, get_profile # Backward-compatible alias for the seed profile allow-list string. @@ -41,6 +46,29 @@ ALLOWED_TOOLS = get_profile("green-commit-only").allowed_tools ToolEventCallback = Callable[[dict[str, Any]], None] +def _kill_process(proc: subprocess.Popen[str] | Any) -> None: + poll = getattr(proc, "poll", None) + if callable(poll): + try: + status = poll() + except Exception: + status = None + if isinstance(status, int): + return + kill = getattr(proc, "kill", None) + if callable(kill): + try: + kill() + except Exception: + return + wait = getattr(proc, "wait", None) + if callable(wait): + try: + wait(timeout=2) + except Exception: + return + + def _is_tool_event(event: dict[str, Any]) -> bool: """True for tool_use/tool_result content blocks and hook lifecycle events. @@ -64,11 +92,13 @@ class AgenticClaudeCodeAdapter(ClaudeCodeAdapter): *, tool_profile: str | ToolProfile = "green-commit-only", on_tool_event: ToolEventCallback | None = None, + cancel: ExecutionCancel | None = None, **kwargs, ): super().__init__(**kwargs) self._workdir = workdir self._on_tool_event = on_tool_event + self._cancel = cancel if isinstance(tool_profile, ToolProfile): self._profile = tool_profile else: @@ -105,27 +135,23 @@ class AgenticClaudeCodeAdapter(ClaudeCodeAdapter): return response def _execute_blocking(self, cmd: list[str], prompt: str, timeout: int) -> LLMResponse: - try: - result = subprocess.run( - cmd, - input=prompt, - capture_output=True, - text=True, - timeout=timeout, - cwd=self._workdir, - ) - except subprocess.TimeoutExpired as exc: - raise LLMTimeoutError( - f"claude CLI timed out after {timeout}s", cause=exc - ) from exc - if result.returncode != 0: + proc = subprocess.Popen( + cmd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + cwd=self._workdir, + ) + stdout, stderr = self._wait_for_process(proc, prompt, timeout) + if proc.returncode != 0: raise LLMSubprocessError( - f"claude CLI exited with code {result.returncode}", - return_code=result.returncode, - stderr=result.stderr, + f"claude CLI exited with code {proc.returncode}", + return_code=proc.returncode, + stderr=stderr, ) return LLMResponse( - content=result.stdout, + content=stdout, model=self._model or "claude-code-cli", usage={}, finish_reason="stop", @@ -170,12 +196,7 @@ class AgenticClaudeCodeAdapter(ClaudeCodeAdapter): proc.stdin.write(prompt) proc.stdin.close() reader_thread.start() - try: - returncode = proc.wait(timeout=timeout) - except subprocess.TimeoutExpired as exc: - proc.kill() - proc.wait() - raise LLMTimeoutError(f"claude CLI timed out after {timeout}s", cause=exc) from exc + returncode = self._wait_for_process(proc, None, timeout, communicate=False) reader_thread.join(timeout=5) stderr = proc.stderr.read() if proc.stderr else "" @@ -200,6 +221,38 @@ class AgenticClaudeCodeAdapter(ClaudeCodeAdapter): }, ) + def _wait_for_process( + self, + proc: subprocess.Popen[str], + prompt: str | None, + timeout: int, + *, + communicate: bool = True, + ) -> Any: + cancel = resolve_cancel(self._cancel) + if cancel is not None: + cancel.check() + cancel.register_process(proc) + try: + if communicate: + result: Any = proc.communicate(input=prompt, timeout=timeout) + else: + result = proc.wait(timeout=timeout) + except subprocess.TimeoutExpired as exc: + _kill_process(proc) + raise LLMTimeoutError( + f"claude CLI timed out after {timeout}s", cause=exc + ) from exc + except ExecutionCancelled: + raise + except Exception: + if cancel is not None and cancel.cancelled: + raise ExecutionCancelled(cancel.reason or "cancelled") from None + raise + if cancel is not None: + cancel.check() + return result + @staticmethod def _handle_stream_event(event: dict[str, Any], text_parts: list[str]) -> None: if event.get("type") != "assistant": diff --git a/rein_aharness/approaches.py b/rein_aharness/approaches.py index 038d372..e8d0642 100644 --- a/rein_aharness/approaches.py +++ b/rein_aharness/approaches.py @@ -17,6 +17,13 @@ from pathlib import Path from typing import Any from rein_aharness.ops_run_client import OpsRun, OpsRunConfig, resolve_ops_target +from rein_aharness.repository_transaction import ( + DirtyRepositoryError, + GitRepositoryError, + RepositoryBusyError, + RepositoryTransaction, + RepositoryTransactionError, +) from rein_aharness.taskspec import TaskSpecError # Approach command names (stable; used in metrics + ops_run.result) @@ -188,7 +195,7 @@ def execute_approach( reopen=False, ) - try: + def _dispatch(tx: RepositoryTransaction | None) -> ApproachResult: if name == APPROACH_FI_RESEARCH_BRIEF: return _run_fi(target, report_to_hub=report_to_hub, commit=commit) if name == APPROACH_BRIEF_DAILY: @@ -204,7 +211,53 @@ def execute_approach( target, report_to_hub=report_to_hub, commit=commit ) if name == APPROACH_AGENT_SESSION: - return _run_agent_session(run, target, report_to_hub=report_to_hub) + return _run_agent_session( + run, target, report_to_hub=report_to_hub, transaction=tx + ) + return ApproachResult( + ok=False, + approach=name, + reason=f"approach not implemented: {name}", + reopen=False, + ) + + if name == APPROACH_AGENT_SESSION: + try: + return _dispatch(None) + except Exception as exc: # noqa: BLE001 — surface as fail ops_run + return ApproachResult( + ok=False, + approach=name, + reason=f"{type(exc).__name__}: {exc}", + reopen=True, + ) + + try: + with RepositoryTransaction(target, correlation_id=run.id) as tx: + result = _dispatch(tx) + result.result["repository_transaction"] = tx.evidence() + return result + except DirtyRepositoryError as exc: + return ApproachResult( + ok=False, + approach=name, + reason=f"refused: {exc}", + reopen=False, + ) + except RepositoryBusyError as exc: + return ApproachResult( + ok=False, + approach=name, + reason=f"refused: {exc}", + reopen=True, + ) + except (GitRepositoryError, RepositoryTransactionError) as exc: + return ApproachResult( + ok=False, + approach=name, + reason=f"refused: {exc}", + reopen=False, + ) except Exception as exc: # noqa: BLE001 — surface as fail ops_run return ApproachResult( ok=False, @@ -213,14 +266,6 @@ def execute_approach( reopen=True, ) - return ApproachResult( - ok=False, - approach=name, - reason=f"approach not implemented: {name}", - reopen=False, - ) - - def _run_fi(target: Path, *, report_to_hub: bool, commit: bool) -> ApproachResult: from rein_aharness.fi_research_brief import run_fi_research_brief @@ -360,7 +405,11 @@ def _run_mail_pipeline( def _run_agent_session( - run: OpsRun, target: Path, *, report_to_hub: bool + run: OpsRun, + target: Path, + *, + report_to_hub: bool, + transaction: RepositoryTransaction | None = None, ) -> ApproachResult: from rein_aharness.ops_run_client import ops_run_to_taskspec from rein_aharness.runner import run_task @@ -384,16 +433,19 @@ def _run_agent_session( ) # target already resolved into TaskSpec assert spec.target_repo == target or True - r = run_task(spec, report_to_hub=report_to_hub) + r = run_task(spec, report_to_hub=report_to_hub, transaction=transaction) + result = { + "committed": r.committed, + "head_after": r.head_after, + "tool_profile": r.tool_profile, + "tokens_spent": r.tokens_spent, + } + if r.transaction: + result["repository_transaction"] = r.transaction return ApproachResult( ok=r.ok, approach=APPROACH_AGENT_SESSION, - result={ - "committed": r.committed, - "head_after": r.head_after, - "tool_profile": r.tool_profile, - "tokens_spent": r.tokens_spent, - }, + result=result, reason=r.reason, reopen=not r.ok, ) diff --git a/rein_aharness/claim_loop.py b/rein_aharness/claim_loop.py index d32f11f..27b51dd 100644 --- a/rein_aharness/claim_loop.py +++ b/rein_aharness/claim_loop.py @@ -23,16 +23,30 @@ from rein_aharness.approaches import ( execute_approach, select_approach, ) +from rein_aharness.execution_cancel import ( + ExecutionCancel, + ExecutionCancelled, + using_cancel, +) from rein_aharness.glas_execution import ( GLAS_APPROACH, GlasExecutionError, execute_profiled_run, ) -from rein_aharness.lease_monitor import LeaseMonitor +from rein_aharness.lease_monitor import LeaseLoss, LeaseMonitor +from rein_aharness.repository_transaction import ( + DirtyRepositoryError, + GitRepositoryError, + RepositoryBusyError, + RepositoryTransaction, + RepositoryTransactionError, +) +from rein_aharness.taskspec import TaskSpecError from rein_aharness.ops_run_client import ( ActivityCoreOpsClient, OpsRun, OpsRunConfig, + resolve_ops_target, OpsRunError, ) @@ -57,19 +71,41 @@ def _heartbeat_interval(lease_seconds: int) -> float: return max(30.0, min(300.0, lease_seconds / 3.0)) +_active_run_cancel: ExecutionCancel | None = None +_active_run_cancel_lock = threading.Lock() + + +def _set_active_run_cancel(cancel: ExecutionCancel | None) -> None: + global _active_run_cancel + with _active_run_cancel_lock: + _active_run_cancel = cancel + + +def _cancel_active_run(reason: str) -> None: + with _active_run_cancel_lock: + cancel = _active_run_cancel + if cancel is not None: + cancel.cancel(reason) + + class _Heartbeat: def __init__( self, client: ActivityCoreOpsClient, run_id: str, lease_seconds: int, + cancel: ExecutionCancel | None = None, ): self._client = client self._run_id = run_id self._lease = lease_seconds self._stop = threading.Event() self._thread: threading.Thread | None = None - self.monitor = LeaseMonitor() + self.cancel = cancel if cancel is not None else ExecutionCancel() + self.monitor = LeaseMonitor(self._on_lost) + + def _on_lost(self, _loss: LeaseLoss) -> None: + self.cancel.cancel("lease-loss") def start(self) -> None: interval = _heartbeat_interval(self._lease) @@ -109,6 +145,41 @@ class _Heartbeat: self.monitor.stop() +def _cancelled_result( + run_id: str, + approach: str | None, + heartbeat: _Heartbeat, +) -> ProcessResult | None: + """Skip Activity Core close after lease loss or execution cancellation.""" + loss = heartbeat.monitor.evidence + if heartbeat.monitor.lost: + assert loss is not None + return ProcessResult( + claimed=True, + run_id=run_id, + approach=approach, + ok=False, + reason=f"lease lost ({loss.error_type})", + detail={ + "lease_loss": { + "error_type": loss.error_type, + "observed_at": loss.observed_at, + } + }, + ) + if heartbeat.cancel.cancelled: + reason = heartbeat.cancel.reason or "cancelled" + return ProcessResult( + claimed=True, + run_id=run_id, + approach=approach, + ok=False, + reason=f"execution cancelled ({reason})", + detail={"cancellation": heartbeat.cancel.evidence()}, + ) + return None + + def process_one( client: ActivityCoreOpsClient | None = None, *, @@ -175,39 +246,30 @@ def process_one( report_to_hub=report_to_hub, ) - hb = _Heartbeat(client, run.id, cfg.lease_seconds) + cancel = ExecutionCancel() + hb = _Heartbeat(client, run.id, cfg.lease_seconds, cancel=cancel) hb.start() execution_error: str | None = None + _set_active_run_cancel(cancel) try: try: - ar: ApproachResult = execute_approach( - run, - approach=approach, - config=cfg, - report_to_hub=report_to_hub, - commit=commit, - ) + with using_cancel(cancel): + ar: ApproachResult = execute_approach( + run, + approach=approach, + config=cfg, + report_to_hub=report_to_hub, + commit=commit, + ) except Exception as exc: execution_error = type(exc).__name__ finally: hb.stop() + _set_active_run_cancel(None) - if hb.monitor.lost: - loss = hb.monitor.evidence - assert loss is not None - return ProcessResult( - claimed=True, - run_id=run.id, - approach=approach, - ok=False, - reason=f"lease lost ({loss.error_type})", - detail={ - "lease_loss": { - "error_type": loss.error_type, - "observed_at": loss.observed_at, - } - }, - ) + refused = _cancelled_result(run.id, approach, hb) + if refused is not None: + return refused if execution_error is not None: return ProcessResult( @@ -280,40 +342,50 @@ def _process_profiled_run( report_to_hub: bool, ) -> ProcessResult: """Execute a profiled row without consulting or falling back to legacy routing.""" - hb = _Heartbeat(client, run.id, client.config.lease_seconds) + 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: - gateway_result = execute_profiled_run( - run, - client.config, - report_to_hub=report_to_hub, - ) + 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) finally: hb.stop() + _set_active_run_cancel(None) - if hb.monitor.lost: - loss = hb.monitor.evidence - assert loss is not None - return ProcessResult( - claimed=True, - run_id=run.id, - approach=GLAS_APPROACH, - ok=False, - reason=f"lease lost ({loss.error_type})", - detail={ - "lease_loss": { - "error_type": loss.error_type, - "observed_at": loss.observed_at, - } - }, - ) + refused = _cancelled_result(run.id, GLAS_APPROACH, hb) + if refused is not None: + return refused if execution_error is not None: reason = execution_reason or f"profiled execution failed ({execution_error})" @@ -321,7 +393,7 @@ def _process_profiled_run( out = client.fail( run.id, error=reason, - reopen=False, + reopen=execution_reopen, result={"ok": False, "approach": GLAS_APPROACH, "reason": reason}, ) except OpsRunError as close_exc: @@ -345,6 +417,10 @@ def _process_profiled_run( 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} try: if ok: out = client.complete(run.id, result=gateway_result) @@ -362,7 +438,7 @@ def _process_profiled_run( approach=GLAS_APPROACH, ok=False, reason=f"close ops_run failed: {exc}; gateway_ok={ok} {reason}", - detail={"execution_evidence": evidence}, + detail=detail, ) return ProcessResult( @@ -372,7 +448,7 @@ def _process_profiled_run( ok=ok, reason="" if ok else reason, ops_state=out.state, - detail={"execution_evidence": evidence}, + detail=detail, ) @@ -423,6 +499,7 @@ def run_claim_loop( def _handle_sig(*_args: Any) -> None: logger.info("shutdown signal received") stop.set() + _cancel_active_run("signal") signal.signal(signal.SIGINT, _handle_sig) signal.signal(signal.SIGTERM, _handle_sig) diff --git a/rein_aharness/execution_cancel.py b/rein_aharness/execution_cancel.py new file mode 100644 index 0000000..2c374a8 --- /dev/null +++ b/rein_aharness/execution_cancel.py @@ -0,0 +1,141 @@ +"""One-shot cancellation for an in-flight adapter call. + +Lease loss, run timeout, and process signals all need the same adapter-side +effect: stop the current subprocess or HTTP client without retaining provider +output. The claim loop owns the reason; adapters only observe and stop. +""" + +from __future__ import annotations + +import contextvars +import threading +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from typing import Any + +_ALLOWED_REASONS = frozenset({"lease-loss", "timeout", "signal"}) +_REASON_LIMIT = 32 + +_active: contextvars.ContextVar["ExecutionCancel | None"] = contextvars.ContextVar( + "rein_aharness_execution_cancel", + default=None, +) + + +class ExecutionCancelled(RuntimeError): + """Bounded proof that an adapter stopped because the run was cancelled.""" + + def __init__(self, reason: str) -> None: + self.reason = _normalize_reason(reason) + super().__init__(f"execution cancelled ({self.reason})") + + +class ExecutionCancel: + """Thread-safe one-shot cancel with optional process/HTTP stop callbacks.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._event = threading.Event() + self._reason: str | None = None + self._stops: list[Callable[[], None]] = [] + + @property + def cancelled(self) -> bool: + return self._event.is_set() + + @property + def reason(self) -> str | None: + with self._lock: + return self._reason + + def wait(self, timeout: float | None = None) -> bool: + return self._event.wait(timeout) + + def check(self) -> None: + if self.cancelled: + raise ExecutionCancelled(self.reason or "cancelled") + + def cancel(self, reason: str) -> str: + """Record the first reason and invoke registered stop callbacks once.""" + callbacks: list[Callable[[], None]] = [] + with self._lock: + if self._reason is None: + self._reason = _normalize_reason(reason) + self._event.set() + callbacks = list(self._stops) + recorded = self._reason + for callback in callbacks: + _invoke_stop(callback) + return recorded + + def register_stop(self, stop: Callable[[], None]) -> None: + """Register a best-effort stopper; invoke immediately if already cancelled.""" + invoke_now = False + with self._lock: + self._stops.append(stop) + invoke_now = self._reason is not None + if invoke_now: + _invoke_stop(stop) + + def register_process(self, proc: Any) -> None: + """Kill an in-flight subprocess. Safe against mock and exited processes.""" + self.register_stop(lambda: _kill_process(proc)) + + def evidence(self) -> dict[str, str]: + if not self.cancelled: + return {} + return {"cancelled": "true", "reason": self.reason or "cancelled"} + + +def active_cancel() -> ExecutionCancel | None: + return _active.get() + + +def resolve_cancel(explicit: ExecutionCancel | None = None) -> ExecutionCancel | None: + return explicit if explicit is not None else active_cancel() + + +@contextmanager +def using_cancel(cancel: ExecutionCancel) -> Iterator[ExecutionCancel]: + token = _active.set(cancel) + try: + yield cancel + finally: + _active.reset(token) + + +def _normalize_reason(reason: str) -> str: + text = (reason or "").strip()[:_REASON_LIMIT] + if text in _ALLOWED_REASONS: + return text + return "cancelled" + + +def _invoke_stop(stop: Callable[[], None]) -> None: + try: + stop() + except Exception: + return + + +def _kill_process(proc: Any) -> None: + poll = getattr(proc, "poll", None) + if callable(poll): + try: + status = poll() + except Exception: + status = None + if isinstance(status, int): + return + kill = getattr(proc, "kill", None) + if callable(kill): + try: + kill() + except Exception: + return + wait = getattr(proc, "wait", None) + if callable(wait): + try: + wait(timeout=2) + except Exception: + return diff --git a/rein_aharness/glas_execution.py b/rein_aharness/glas_execution.py index 6afe366..5bcdbda 100644 --- a/rein_aharness/glas_execution.py +++ b/rein_aharness/glas_execution.py @@ -11,6 +11,7 @@ from __future__ import annotations from collections.abc import Callable from typing import Any +from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled, resolve_cancel from rein_aharness.ops_run_client import OpsRun, OpsRunConfig, resolve_ops_target GLAS_APPROACH = "glas-profile" @@ -65,8 +66,12 @@ def execute_profiled_run( report_to_hub: bool = True, request_factory: Callable[..., Any] | None = None, gateway: Callable[[Any], Any] | None = None, + cancel: ExecutionCancel | None = None, ) -> dict[str, Any]: """Invoke Glas and return its complete JSON-compatible GatewayResult.""" + guard = resolve_cancel(cancel) + if guard is not None: + guard.check() if request_factory is None or gateway is None: try: from glas_harness.contract import ExecutionRequest @@ -83,10 +88,16 @@ def execute_profiled_run( request = request_factory(**_request_kwargs(run, config, report_to_hub)) result = gateway(request) raw = result.model_dump(mode="json") if hasattr(result, "model_dump") else result + except ExecutionCancelled: + raise except GlasExecutionError: raise except Exception as exc: + if guard is not None and guard.cancelled: + raise ExecutionCancelled(guard.reason or "cancelled") from None raise GlasExecutionError(f"Glas gateway invocation failed: {exc}") from exc + if guard is not None: + guard.check() if not isinstance(raw, dict) or not isinstance(raw.get("ok"), bool): raise GlasExecutionError("Glas gateway returned an invalid GatewayResult") diff --git a/rein_aharness/llm_connect_client.py b/rein_aharness/llm_connect_client.py index ad7faf5..2ca34dc 100644 --- a/rein_aharness/llm_connect_client.py +++ b/rein_aharness/llm_connect_client.py @@ -11,6 +11,12 @@ from typing import Any import httpx +from rein_aharness.execution_cancel import ( + ExecutionCancel, + ExecutionCancelled, + resolve_cancel, +) + _SAFE_RESPONSE_METADATA_KEYS = frozenset( { "finish_reason", @@ -50,20 +56,27 @@ class LLMConnectClient: *, model: str = "", config: dict[str, Any] | None = None, + cancel: ExecutionCancel | None = None, ) -> str: run_config = dict(config or {}) if model and "model_name" not in run_config: run_config["model_name"] = model run_config.setdefault("timeout_seconds", int(self.timeout_seconds)) payload: dict[str, Any] = {"prompt": prompt, "config": run_config} + url = f"{self.base_url}/execute" + guard = resolve_cancel(cancel) + if guard is not None: + guard.check() try: - resp = httpx.post( - f"{self.base_url}/execute", - json=payload, - timeout=self.timeout_seconds, - ) + resp = self._post(url, payload, guard) + except ExecutionCancelled: + raise except httpx.HTTPError as exc: + if guard is not None and guard.cancelled: + raise ExecutionCancelled(guard.reason or "cancelled") from None raise LLMConnectError(f"llm-connect request failed: {exc}") from exc + if guard is not None: + guard.check() if resp.status_code >= 400: raise LLMConnectError(_llm_connect_error_text(resp)) try: @@ -76,6 +89,18 @@ class LLMConnectClient: raise LLMConnectError("llm-connect response missing string content") return content + def _post( + self, + url: str, + payload: dict[str, Any], + guard: ExecutionCancel | None, + ) -> httpx.Response: + if guard is None: + return httpx.post(url, json=payload, timeout=self.timeout_seconds) + with httpx.Client(timeout=self.timeout_seconds) as client: + guard.register_stop(client.close) + return client.post(url, json=payload) + def _llm_connect_error_text(resp: httpx.Response) -> str: """Keep llm-connect's safe cause without copying an upstream response blob.""" diff --git a/rein_aharness/repository_transaction.py b/rein_aharness/repository_transaction.py index c913eb4..e3889ec 100644 --- a/rein_aharness/repository_transaction.py +++ b/rein_aharness/repository_transaction.py @@ -1,8 +1,9 @@ """Process-safe transaction boundary for a local Git checkout. -This module is intentionally not wired into the claim loop yet. It provides -the repository-local half of HARNESS-WP-0003-T02 while the Activity Core lease -boundary is still under owner review. +Wired into `run_task`, profile-absent `execute_approach` mutators, and the +profiled claim path under REINAH-WP-0003-T02 / ADR-002. Repository +acceptance (T03) remains a separate read-only validator and is not applied +to live results yet. """ from __future__ import annotations diff --git a/rein_aharness/runner.py b/rein_aharness/runner.py index 86d1f5b..4febd55 100644 --- a/rein_aharness/runner.py +++ b/rein_aharness/runner.py @@ -17,9 +17,17 @@ from pathlib import Path from typing import Any, Callable from rein_aharness import hub, metrics +from rein_aharness.execution_cancel import ExecutionCancelled from rein_aharness.manifest import resolve_run_policy from rein_aharness.persona import load_persona_bundle from rein_aharness.profiles import UnknownToolProfileError, get_profile +from rein_aharness.repository_transaction import ( + DirtyRepositoryError, + GitRepositoryError, + RepositoryBusyError, + RepositoryTransaction, + RepositoryTransactionError, +) from rein_aharness.taskspec import TaskSpec PROMPT_TEMPLATE = """\ @@ -59,6 +67,7 @@ class RunResult: # called with emit_tool_events=True. See adapter.py's module docstring # for why this is observation, not external tool dispatch. tool_events: list[dict[str, Any]] = field(default_factory=list) + transaction: dict[str, Any] | None = None def _git(repo: Path, *args: str) -> str: @@ -73,6 +82,28 @@ def _git(repo: Path, *args: str) -> str: return result.stdout.strip() +def _refused_run( + *, + reason: str, + model: str | None = None, + head_before: str = "", + transaction: dict[str, Any] | None = None, +) -> RunResult: + return RunResult( + ok=False, + committed=False, + head_before=head_before, + head_after=head_before, + persona_source="none", + session_output="", + reason=reason, + tool_profile="", + budget_tokens=None, + model=model, + transaction=transaction, + ) + + def run_task( spec: TaskSpec, adapter=None, @@ -83,21 +114,14 @@ def run_task( model: str | None = None, tool_profile_override: str | None = None, budget_tokens_override: int | None = None, + transaction: RepositoryTransaction | None = None, ) -> RunResult: if spec.repository_grant is not None: - return RunResult( - ok=False, - committed=False, - head_before="", - head_after="", - persona_source="none", - session_output="", + return _refused_run( reason=( "refused: repository_grant enforcement is not enabled; " "no adapter was dispatched" ), - tool_profile="", - budget_tokens=None, model=model, ) try: @@ -110,31 +134,9 @@ def run_task( budget_tokens = budget_tokens_override profile = get_profile(profile_name) except UnknownToolProfileError as exc: - return RunResult( - ok=False, - committed=False, - head_before="", - head_after="", - persona_source="none", - session_output="", - reason=f"refused: {exc}", - tool_profile="", - budget_tokens=None, - model=model, - ) + return _refused_run(reason=f"refused: {exc}", model=model) except Exception as exc: - return RunResult( - ok=False, - committed=False, - head_before="", - head_after="", - persona_source="none", - session_output="", - reason=f"manifest resolution failed: {exc}", - tool_profile="", - budget_tokens=None, - model=model, - ) + return _refused_run(reason=f"manifest resolution failed: {exc}", model=model) collected_events: list[dict[str, Any]] = [] @@ -162,7 +164,58 @@ def run_task( adapter_kwargs["model"] = model adapter = AgenticClaudeCodeAdapter(**adapter_kwargs) - head_before = _git(spec.target_repo, "rev-parse", "HEAD") + def _run_locked(tx: RepositoryTransaction) -> RunResult: + return _execute_locked_task( + spec, + tx, + adapter=adapter, + profile=profile, + blueprint=blueprint, + lane=lane, + budget_tokens=budget_tokens, + collected_events=collected_events, + report_to_hub=report_to_hub, + write_metrics=write_metrics, + model=model, + ) + + if transaction is not None: + return _run_locked(transaction) + try: + with RepositoryTransaction( + spec.target_repo, + correlation_id=spec.hub_task_id or spec.title, + ) as tx: + return _run_locked(tx) + except DirtyRepositoryError as exc: + return _refused_run( + reason=f"refused: {exc}", + model=model, + head_before=exc.baseline.head, + transaction={"baseline": exc.baseline.evidence()}, + ) + except RepositoryBusyError as exc: + return _refused_run(reason=f"refused: {exc}", model=model) + except (GitRepositoryError, RepositoryTransactionError) as exc: + return _refused_run(reason=f"refused: {exc}", model=model) + + +def _execute_locked_task( + spec: TaskSpec, + tx: RepositoryTransaction, + *, + adapter: Any, + profile: Any, + blueprint: str, + lane: str | None, + budget_tokens: int | None, + collected_events: list[dict[str, Any]], + report_to_hub: bool, + write_metrics: bool, + model: str | None, +) -> RunResult: + assert tx.baseline is not None + head_before = tx.baseline.head persona, persona_source = load_persona_bundle(blueprint, spec.target_repo) prompt = PROMPT_TEMPLATE.format( persona=persona or "(no persona bundle available for this run)", @@ -188,6 +241,11 @@ def run_task( resolved_model = response.model session_ok = True reason = "" + except ExecutionCancelled as exc: + session_output = "" + session_ok = False + reason = f"execution cancelled ({exc.reason})" + resolved_model = model except Exception as exc: # adapter / budget failures must still be reported session_output = "" session_ok = False @@ -217,6 +275,7 @@ def run_task( tokens_spent=tokens_spent, execution_time_s=execution_time_s, tool_events=collected_events, + transaction=tx.evidence(), ) if write_metrics: diff --git a/tests/test_adapter.py b/tests/test_adapter.py index cb4a528..01dc7b8 100644 --- a/tests/test_adapter.py +++ b/tests/test_adapter.py @@ -111,11 +111,58 @@ def test_execute_streaming_raises_timeout(tmp_path) -> None: def test_execute_blocking_path_unchanged_without_callback(tmp_path) -> None: adapter = AgenticClaudeCodeAdapter(workdir=tmp_path) config = RunConfig(timeout_seconds=30) - fake_result = MagicMock(returncode=0, stdout="plain output", stderr="") + proc = MagicMock() + proc.communicate.return_value = ("plain output", "") + proc.returncode = 0 + proc.poll.return_value = 0 - with patch("rein_aharness.adapter.subprocess.run", return_value=fake_result) as run: + with patch("rein_aharness.adapter.subprocess.Popen", return_value=proc) as popen: response = adapter.execute_prompt("do it", config) assert response.content == "plain output" - argv = run.call_args.args[0] + argv = popen.call_args.args[0] assert "--output-format" not in argv + proc.communicate.assert_called_once() + + +def test_blocking_cancel_kills_process_and_raises(tmp_path) -> None: + from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled + + cancel = ExecutionCancel() + adapter = AgenticClaudeCodeAdapter(workdir=tmp_path, cancel=cancel) + config = RunConfig(timeout_seconds=30) + proc = MagicMock() + proc.poll.return_value = None + + def _communicate(input=None, timeout=None): # noqa: A002 + cancel.cancel("lease-loss") + raise RuntimeError("subprocess interrupted") + + proc.communicate.side_effect = _communicate + + with patch("rein_aharness.adapter.subprocess.Popen", return_value=proc): + with pytest.raises(ExecutionCancelled, match="lease-loss"): + adapter.execute_prompt("do it", config) + proc.kill.assert_called_once() + + +def test_streaming_cancel_refuses_nonzero_exit(tmp_path) -> None: + from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled + + cancel = ExecutionCancel() + adapter = AgenticClaudeCodeAdapter( + workdir=tmp_path, on_tool_event=lambda e: None, cancel=cancel + ) + config = RunConfig(timeout_seconds=30) + proc = _fake_proc([]) + proc.poll.return_value = None + + def _wait(timeout=None): + cancel.cancel("signal") + return 1 + + proc.wait.side_effect = _wait + + with patch("rein_aharness.adapter.subprocess.Popen", return_value=proc): + with pytest.raises(ExecutionCancelled, match="signal"): + adapter.execute_prompt("do it", config) diff --git a/tests/test_claim_loop.py b/tests/test_claim_loop.py index 40c3910..8eb841c 100644 --- a/tests/test_claim_loop.py +++ b/tests/test_claim_loop.py @@ -6,7 +6,13 @@ from unittest.mock import MagicMock, patch import time from rein_aharness.approaches import ApproachResult, APPROACH_FI_RESEARCH_BRIEF -from rein_aharness.claim_loop import process_one, poll_peek +from rein_aharness.claim_loop import ( + _cancel_active_run, + _set_active_run_cancel, + process_one, + poll_peek, +) +from rein_aharness.execution_cancel import ExecutionCancel, active_cancel from rein_aharness.glas_execution import GLAS_APPROACH, GlasExecutionError from rein_aharness.ops_run_client import ( ActivityCoreOpsClient, @@ -135,6 +141,78 @@ def test_process_one_refuses_close_after_lease_loss() -> None: client.fail.assert_not_called() +def test_process_one_lease_loss_cancels_registered_adapter_process() -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + client.claim.return_value = [_claimed_run()] + client.heartbeat.side_effect = OpsRunError("lease rejected", status_code=409) + proc = MagicMock() + proc.poll.return_value = None + + def slow_execute(*_args, **_kwargs): + cancel = active_cancel() + assert cancel is not None + cancel.register_process(proc) + deadline = time.monotonic() + 1.0 + while not cancel.cancelled and time.monotonic() < deadline: + time.sleep(0.005) + cancel.check() + + with ( + patch("rein_aharness.claim_loop._heartbeat_interval", return_value=0.01), + patch("rein_aharness.claim_loop.execute_approach", side_effect=slow_execute), + ): + result = process_one(client) + + assert result.ok is False + assert result.reason.startswith("lease lost") + proc.kill.assert_called() + client.complete.assert_not_called() + client.fail.assert_not_called() + + +def test_process_one_skips_close_after_signal_cancel() -> None: + client = MagicMock(spec=ActivityCoreOpsClient) + client.config = OpsRunConfig(worker_id="w", lease_seconds=90) + client.claim.return_value = [_claimed_run()] + ar = ApproachResult( + ok=True, + approach=APPROACH_FI_RESEARCH_BRIEF, + result={"path": "briefs/x.md"}, + reason="ok", + ) + + def execute_then_signal(*_args, **_kwargs): + cancel = active_cancel() + assert cancel is not None + cancel.cancel("signal") + return ar + + with patch("rein_aharness.claim_loop.execute_approach", side_effect=execute_then_signal): + result = process_one(client) + + assert result.ok is False + assert result.reason == "execution cancelled (signal)" + assert result.detail == {"cancellation": {"cancelled": "true", "reason": "signal"}} + client.complete.assert_not_called() + client.fail.assert_not_called() + + +def test_shutdown_signal_cancels_active_run() -> None: + cancel = ExecutionCancel() + proc = MagicMock() + proc.poll.return_value = None + cancel.register_process(proc) + _set_active_run_cancel(cancel) + try: + _cancel_active_run("signal") + finally: + _set_active_run_cancel(None) + + assert cancel.reason == "signal" + proc.kill.assert_called_once() + + def test_process_one_records_adapter_exception_without_unbound_result() -> None: client = MagicMock(spec=ActivityCoreOpsClient) client.config = OpsRunConfig(worker_id="w", lease_seconds=90) @@ -220,8 +298,16 @@ def test_profiled_run_uses_glas_and_completes_with_full_result() -> None: assert result.ok is True assert result.approach == GLAS_APPROACH - assert result.detail == {"execution_evidence": gateway_result["evidence"]} - client.complete.assert_called_once_with(run.id, result=gateway_result) + assert result.detail["execution_evidence"] == gateway_result["evidence"] + transaction = result.detail["repository_transaction"] + assert transaction["correlation_id"] == run.id + assert transaction["baseline"]["clean"] is True + client.complete.assert_called_once() + 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["repository_transaction"] == transaction client.fail.assert_not_called() select.assert_not_called() execute.assert_not_called() diff --git a/tests/test_execution_cancel.py b/tests/test_execution_cancel.py new file mode 100644 index 0000000..7bac27e --- /dev/null +++ b/tests/test_execution_cancel.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import threading +import time +from unittest.mock import MagicMock + +import pytest + +from rein_aharness.execution_cancel import ( + ExecutionCancel, + ExecutionCancelled, + active_cancel, + using_cancel, +) + + +def test_cancel_is_one_shot_and_bounded() -> None: + cancel = ExecutionCancel() + first = cancel.cancel("lease-loss") + second = cancel.cancel("timeout") + + assert first == second == "lease-loss" + assert cancel.cancelled is True + assert cancel.reason == "lease-loss" + assert cancel.evidence() == {"cancelled": "true", "reason": "lease-loss"} + + +def test_unknown_reason_is_normalized() -> None: + cancel = ExecutionCancel() + cancel.cancel("provider exploded with secrets") + assert cancel.reason == "cancelled" + + +def test_check_raises_only_after_cancel() -> None: + cancel = ExecutionCancel() + cancel.check() + cancel.cancel("timeout") + with pytest.raises(ExecutionCancelled, match="timeout") as excinfo: + cancel.check() + assert excinfo.value.reason == "timeout" + + +def test_register_process_is_killed_on_cancel() -> None: + proc = MagicMock() + proc.poll.return_value = None + cancel = ExecutionCancel() + cancel.register_process(proc) + cancel.cancel("lease-loss") + proc.kill.assert_called_once() + + +def test_already_cancelled_register_kills_immediately() -> None: + proc = MagicMock() + proc.poll.return_value = None + cancel = ExecutionCancel() + cancel.cancel("signal") + cancel.register_process(proc) + proc.kill.assert_called_once() + + +def test_exited_process_is_not_killed() -> None: + proc = MagicMock() + proc.poll.return_value = 0 + cancel = ExecutionCancel() + cancel.register_process(proc) + cancel.cancel("timeout") + proc.kill.assert_not_called() + + +def test_stop_callback_failure_does_not_block_cancel() -> None: + cancel = ExecutionCancel() + cancel.register_stop(lambda: (_ for _ in ()).throw(RuntimeError("stop failed"))) + assert cancel.cancel("signal") == "signal" + + +def test_concurrent_cancel_invokes_stop_once() -> None: + seen = [] + cancel = ExecutionCancel() + cancel.register_stop(lambda: seen.append("stop")) + threads = [ + threading.Thread(target=cancel.cancel, args=("lease-loss",)) for _ in range(8) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert seen == ["stop"] + + +def test_using_cancel_exposes_active_cancel() -> None: + cancel = ExecutionCancel() + assert active_cancel() is None + with using_cancel(cancel): + assert active_cancel() is cancel + cancel.cancel("timeout") + assert active_cancel() is cancel + assert active_cancel() is None + + +def test_wait_returns_after_cancel() -> None: + cancel = ExecutionCancel() + started = threading.Event() + + def _cancel_soon() -> None: + started.wait(timeout=1) + time.sleep(0.01) + cancel.cancel("signal") + + thread = threading.Thread(target=_cancel_soon) + thread.start() + started.set() + assert cancel.wait(timeout=1) is True + thread.join() diff --git a/tests/test_glas_execution.py b/tests/test_glas_execution.py index 76a5e78..32c2616 100644 --- a/tests/test_glas_execution.py +++ b/tests/test_glas_execution.py @@ -127,3 +127,48 @@ def test_profiled_actor_validates_against_real_glas_and_sandboxer(tmp_path: Path assert captured["request"].actor == "agt" assert captured["sandbox_request"].consumer.actor == sandbox_models.ActorType.AGT assert captured["sandbox_request"].consumer.run_id == "run-1" + + +def test_profiled_run_does_not_invoke_gateway_when_already_cancelled( + tmp_path: Path, +) -> None: + from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled + + repo = _repo(tmp_path) + cancel = ExecutionCancel() + cancel.cancel("lease-loss") + called = [] + with pytest.raises(ExecutionCancelled, match="lease-loss"): + execute_profiled_run( + _run(repo), + OpsRunConfig(repo_roots=(str(tmp_path),)), + request_factory=lambda **kwargs: kwargs, + gateway=lambda request: called.append(request) or { + "ok": True, + "evidence": {"outcome": "succeeded"}, + }, + cancel=cancel, + ) + assert called == [] + + +def test_profiled_run_refuses_success_if_cancelled_during_gateway( + tmp_path: Path, +) -> None: + from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled + + repo = _repo(tmp_path) + cancel = ExecutionCancel() + + def gateway(request): + cancel.cancel("timeout") + return {"ok": True, "evidence": {"outcome": "succeeded"}} + + with pytest.raises(ExecutionCancelled, match="timeout"): + execute_profiled_run( + _run(repo), + OpsRunConfig(repo_roots=(str(tmp_path),)), + request_factory=lambda **kwargs: kwargs, + gateway=gateway, + cancel=cancel, + ) diff --git a/tests/test_mail_triage.py b/tests/test_mail_triage.py index c2166da..4f7c9f9 100644 --- a/tests/test_mail_triage.py +++ b/tests/test_mail_triage.py @@ -257,3 +257,31 @@ def test_get_client_requires_env(monkeypatch) -> None: from rein_aharness.llm_connect_client import get_llm_connect_client get_llm_connect_client() + + +def test_llm_connect_cancel_closes_client_and_raises(monkeypatch) -> None: + import rein_aharness.llm_connect_client as mod + from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled + + closed = [] + cancel = ExecutionCancel() + + class FakeClient: + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def close(self): + closed.append("closed") + + def post(self, url, json=None): # noqa: A002 + cancel.cancel("timeout") + raise mod.httpx.ConnectError("connection reset") + + monkeypatch.setattr(mod.httpx, "Client", lambda timeout=None: FakeClient()) + client = LLMConnectClient("http://llm.test", timeout_seconds=5) + with pytest.raises(ExecutionCancelled, match="timeout"): + client.complete("hi", cancel=cancel) + assert closed == ["closed"] diff --git a/tests/test_runner.py b/tests/test_runner.py index 89ef869..3376a3e 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -126,6 +126,28 @@ def test_run_task_fails_without_commit(tmp_path) -> None: assert result.reason == "session completed without committing" +def test_run_task_records_bounded_cancellation_without_session_output(tmp_path) -> None: + from rein_aharness.execution_cancel import ExecutionCancelled + + repo = _make_repo(tmp_path) + + class CancellingAdapter: + def execute_prompt(self, prompt, config): + raise ExecutionCancelled("lease-loss") + + result = run_task( + _spec(repo), + adapter=CancellingAdapter(), + report_to_hub=False, + write_metrics=False, + ) + + assert result.ok is False + assert result.committed is False + assert result.session_output == "" + assert result.reason == "execution cancelled (lease-loss)" + + def test_run_task_refuses_unknown_tool_profile(tmp_path) -> None: repo = _make_repo(tmp_path) _write_manifest( @@ -160,6 +182,21 @@ def test_run_task_resolves_manifest_profile_and_budget(tmp_path) -> None: } }, ) + subprocess.run(["git", "add", ".kaizen"], cwd=repo, check=True) + subprocess.run( + [ + "git", + "-c", + "user.email=t@t", + "-c", + "user.name=t", + "commit", + "-qm", + "configure rein", + ], + cwd=repo, + check=True, + ) adapter = CommittingAdapter(repo) result = run_task( @@ -186,6 +223,21 @@ def test_run_task_budget_exhaustion_fails(tmp_path) -> None: } }, ) + subprocess.run(["git", "add", ".kaizen"], cwd=repo, check=True) + subprocess.run( + [ + "git", + "-c", + "user.email=t@t", + "-c", + "user.name=t", + "commit", + "-qm", + "configure rein", + ], + cwd=repo, + check=True, + ) result = run_task( _spec(repo), diff --git a/workplans/REINAH-WP-0003-governed-runtime-integrity.md b/workplans/REINAH-WP-0003-governed-runtime-integrity.md index 9d95fad..39ba5c0 100644 --- a/workplans/REINAH-WP-0003-governed-runtime-integrity.md +++ b/workplans/REINAH-WP-0003-governed-runtime-integrity.md @@ -9,7 +9,7 @@ owner: codex topic_slug: rein-aharness priority: high created: "2026-08-23" -updated: "2026-08-23" +updated: "2026-09-04" related: - REIN-A-0004 - GLAS-IN-0002 @@ -65,7 +65,7 @@ waiting on the sandbox owner. ```task id: REINAH-WP-0003-T01 -status: progress +status: done priority: high state_hub_task_id: "596fc138-1ee3-5602-95c7-0bd728ec70ab" ``` @@ -140,6 +140,19 @@ and T02–T06 can cite stable decisions rather than infer ownership from code. sand-boxer (`1bf5768c-f77a-4f08-ae02-25caff9f0166`), llm-connect (`4bd35ce5-9c84-4695-83d5-c9bd16fa1b77`), and Glas (`52567711-1100-43c0-9f8e-40650738eb04`). +- 2026-08-29: this State Hub instance no longer serves those request + message ids (GET 404). Glas approval of the earlier revision still + stands in the ADR text; Activity Core, sand-boxer, and llm-connect + exact-revision acknowledgements of `36e1096` / + `84b47d2e…bb206f` remain the T01 gate. Reminder messages resent from + `rein-aharness`: Activity Core `92b80a06-dc0e-46bd-a5c7-b43d1babe9ba`, + Glas `afc2af7b-8246-482b-a6f5-7acfc2aa32b1`, sand-boxer + `b220425d-5295-4bf3-af4c-34ec838f1175`, llm-connect + `4cc4a2cc-584b-423d-8f34-a0bcd981f1c7`. +- 2026-08-29: operator Bernd Worsch accepted ADR-002 as written. + `INTENT.md` and ADR-001 slogan wording were amended the same day. + Sibling exact-revision acknowledgements remain welcome as recording, + not a gate. T01 is `done`; T04 and T05 are unblocked. ## Make each repository run a lease-bound transaction @@ -211,6 +224,40 @@ class marker rather than leaving an unbound result or retaining provider text. explicit ownership-rejection statuses as lease loss while allowing transport errors to retry without immediately abandoning a run. +### Adapter cancellation slice — 2026-08-29 + +Connected lease-loss, timeout, and signal cancellation at the adapter +boundaries and enabled the repository transaction after ADR-002 acceptance. + +- `rein_aharness/execution_cancel.py` is a one-shot, adapter-neutral cancel + token. It records a bounded reason (`lease-loss` / `timeout` / `signal`), + invokes registered process/HTTP stoppers once, and is published to adapters + through an explicit argument or `using_cancel()` context. +- The claim-loop heartbeat now cancels that token when Activity Core rejects + ownership. SIGINT/SIGTERM also cancel the active run. In both cases the loop + still skips Activity Core complete/fail and returns bounded cancellation or + lease-loss evidence. +- `AgenticClaudeCodeAdapter` waits on `Popen` for both blocking and streaming + sessions so an in-flight Claude process can be killed. `LLMConnectClient` + closes its HTTP client. Glas checks the token before and after + `run_execution` and refuses a late success; it still cannot abort an + in-flight sandbox (sand-boxer residual). +- `run_task` records `execution cancelled ()` without session output. + Tenant brief executors pick the token up through the active-cancel context + when they call llm-connect. + +The repository lock is now wired into `runner.py`, legacy mutating approaches, +and the profiled claim path following operator acceptance of ADR-002. Remaining +T02 work is to prove lock release on timeout, signal, and result-close failure +and close the task with the resulting bounded evidence. + +### Review checkpoint — 2026-09-04 + +Reviewed the cancellation and transaction wiring together and corrected the +affected profiled-result and clean-baseline test setup. Full verification passes: +`209 passed, 1 skipped`. T02 remains `progress` until the named timeout, signal, +and result-close lock-release cases are covered directly. + ## Verify accepted commits and reconcile metrics/reporting ```task @@ -328,7 +375,7 @@ was requested. ```task id: REINAH-WP-0003-T04 -status: wait +status: todo priority: high state_hub_task_id: "f6c7f80c-3807-54ac-b44a-ef582433b927" ``` @@ -364,7 +411,7 @@ This is read-only preparation; T04 remains `wait` until T01 accepts ADR-002. ```task id: REINAH-WP-0003-T05 -status: wait +status: todo priority: high state_hub_task_id: "c4a3f08f-2874-5672-9f29-aecddd697d90" ```