From 3a52be15a056fadc6dc5ccfa3d348b5b380cf6b7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:36:24 +0800 Subject: [PATCH 1/4] fix(turn): observe PR wait dependencies before business admission Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/agent-loop-effect-interpreter-v0.md | 13 + .../agent-loop-effect-interpreter-v0.zh-CN.md | 8 + docs/reference/todo-continuation-readback.md | 35 +- loopx/cli_commands/turn.py | 47 +++ loopx/cli_commands/turn_rendering.py | 34 +- .../capabilities/test_pr_wait_observation.py | 370 ++++++++++++++++++ 6 files changed, 503 insertions(+), 4 deletions(-) diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md index 7966e9f4d5..8b5724aa3b 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md @@ -471,6 +471,19 @@ shape merely because of its size. Optional memory participation now has one compact, signed envelope projection; the Codex CLI adapter preserves default-off isolation. This does not qualify installed heartbeat/App adoption or model value. +Fresh executing native Turns now compose the existing opt-in PR dependency +observer before their governing decision. An exact merged observation can +satisfy the original wait without asking the blocked business Host to select +that observation. The TypeScript plan, retained merge evidence and ordinary +execution admission remain the owners; the CLI reports observation cost and +failure/uncertain-write readback. File/SQLite qualification covers paused +business compute, subsequent validated protocol-host work and original-journal +replay, with the original requirement, ownership and merge timestamp retained. +Plans, dry runs, missing declarations and journal recovery add no observation. +This is a native CLI adoption slice; installed heartbeat declarations, +App/Lark caller adoption, sustained live scheduling, model outcomes and the +three convergence acceptances remain open. + Bounded planning context now remains inline instead of forcing a full-decision read under envelope budget pressure. Its existing owner retains the 3 KiB bound; the shared DSH/cloud adapter transports the signed contribution and, for saved diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md index 5121102100..fd2b0dd3e8 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md @@ -329,6 +329,14 @@ R5 短包投影也完整保留已有 CLI 结算计划,包括 effect identity envelope 投影,Codex CLI 保持关闭隔离;这不代表安装态 heartbeat/App 或模型收益 已验收。 +新的原生 Turn 执行入口在 governing decision 前组合既有 opt-in PR 依赖观察器, +精确合并事实可满足原等待,不要求被该等待阻挡的业务 Host 主动选择观察工作。目标计划、 +保留的合并证据与正常执行准入继续归 TS 及原 owner;CLI 读回观察成本、失败与不确定 +写入。File/SQLite 验证覆盖业务额度暂停、随后经过独立校验的协议宿主工作和原 journal +重放,保留原要求、归属及合并时间。plan、dry run、缺少声明与 journal 恢复不添加观察。 +本切片只完成原生 CLI 接入;安装态 heartbeat 声明、App/Lark 调用方采用、持续真实调度、 +模型结果与以下三批收敛验收保持开放。 + 有界规划上下文现在保持内联,不再因短包预算压力而强制读取整份决策;既有 owner 的 3 KiB 限制保留。共享 DSH/cloud adapter 传递签名内容,历史引用同时保留原捕获路由和 哈希。thin renderer 仅在存在 `agent_context.detail_ref` 时要求能力补读,不能把也用于 diff --git a/docs/reference/todo-continuation-readback.md b/docs/reference/todo-continuation-readback.md index 62fc92e9b4..9207e1e710 100644 --- a/docs/reference/todo-continuation-readback.md +++ b/docs/reference/todo-continuation-readback.md @@ -37,8 +37,24 @@ alone do not opt in. Omitting that declaration keeps external observation off. Explicit prequota invocation authorizes compact metadata reads and local observation writeback, never merge, review, remote mutation or quota spend. Read back with `todo list --goal-id GOAL --todo-id TODO --format json`. -These same typed dependency facts are the boundary to reuse in TurnEnvelope; -this heartbeat adoption does not claim that every Turn host already polls them. + +Fresh `turn run-once --execute` invocations with that same explicit capability +now call the existing PR dependency observer before collecting the governing +decision. Observation can satisfy the wait while business compute is paused; +the ordinary stop, quota, claim, lease and launch checks still govern execution. +Planning, dry runs and `--resume-turn-key` recovery perform no new observation. +Omitting the capability retains the previous path. Disabling the declaration +stops future automatic reads and retains previously verified merge facts. + +The native CLI returns `pr_dependency_observation` in JSON: remote read count, +observer duration in milliseconds, degraded/failure readback, the existing +target/cadence plan and observation `state_written`. Markdown summarizes those +counts, cadence and failures. Observation `state_written` is `null` after an uncertain provider/log failure; a failed response must +not claim zero local effects. Observation has no separate model call or business +quota debit. The existing four-target bound and 30-minute retry cadence apply; +actual network latency and total host costs remain workload-dependent. This +qualifies the native CLI bridge. Installed heartbeat declarations, App/Lark +caller adoption and sustained live scheduling remain separate boundaries. `pr_merged:owner/repo#42` 只标识依赖;缺少合并事件不证明 GitHub 仍在等待。 历史交付监督不再豁免这种未观察等待,因果阻塞结算仍要求已注册的 @@ -56,7 +72,20 @@ provider 失败不能解除等待。观察不修改 Todo 状态、要求、认 prequota;只声明 filesystem/shell 不会启用。省略该声明即关闭外部观察。 显式调用仅授权紧凑元数据读取和本地观察记账,不授权合并、review、远端写入或额度支出。 通过 `todo list --goal-id GOAL --todo-id TODO --format json` 回读。 -TurnEnvelope 应复用同一类型化依赖事实;本批未证明所有 Turn host 已接入自动核验。 + +新的 `turn run-once --execute` 调用在显式声明同一能力后,会在收集本次 governing +decision 前运行既有 PR 依赖观察器。业务额度暂停时仍可满足依赖条件;停止、quota、 +claim、lease 与宿主启动仍由原准入约束。plan、dry run 与 `--resume-turn-key` +恢复不执行新观察;省略能力声明保持原路径。移除声明会停止之后的自动读取,已验证的 +合并事实保留。 + +原生 CLI 的 JSON 返回 `pr_dependency_observation`:远端读取次数、观察耗时毫秒、 +退化/失败读回、既有目标/节奏计划与观察 `state_written`;Markdown 汇总次数、 +节奏与失败。provider +或日志响应不确定时,最后一项为 `null`,失败不能被解释为零本地效果。观察没有单独的 +模型调用或业务额度扣费;沿用每次四项和 30 分钟重试节奏,实际网络延迟及完整宿主成本 +仍依赖工作负载。本切片验证原生 CLI 接入;安装态 heartbeat 声明、App/Lark 调用方采用 +与持续真实调度仍分别验收。 ## Completion dependency cycles / 完成依赖环 diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index c6c05e1684..29dd448f99 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -2,6 +2,7 @@ import argparse import shlex +import time from collections.abc import Callable, Mapping from pathlib import Path from typing import Any @@ -40,6 +41,20 @@ FormatSelector = Callable[..., str] +def _attach_pr_dependency_observation( + payload: dict[str, Any], observation: dict[str, Any] | None, +) -> None: + if observation is None: + return + payload["pr_dependency_observation"] = observation + written = observation["state_written"] + effects = payload.setdefault("effects", {}) + if written is True or (written is None and effects.get("state_written") is False): + effects["state_written"] = written + if written is not False and isinstance(payload.get("boundary"), dict): + payload["boundary"]["read_only"] = False + + def handle_turn_command( args: argparse.Namespace, *, @@ -67,6 +82,7 @@ def handle_turn_command( output_format=output_format, print_payload=print_payload, ) payload: dict[str, Any] = {} + pr_dependency_observation: dict[str, Any] | None = None try: if getattr(args, "todo_id", None) is not None and ( getattr(args, "resume_turn_key", None) @@ -175,6 +191,7 @@ def handle_turn_command( from ..capabilities.agent_turn_recall import run_configured_agent_turn_recall_fail_open from ..capabilities.periodic_report.cadence_runtime import extend_cadence_turn_start_dispatch from .lark_inbox import dispatch_goal_lark_turn_start_hooks + from ..control_plane.todos.contract import normalize_required_capabilities turn_start_hook_dispatch = dispatch_goal_lark_turn_start_hooks( registry_path=registry_path, @@ -197,6 +214,34 @@ def handle_turn_command( agent_id=args.agent_id, available=args.available_capabilities, ) + + # Observe synchronized dependencies before selection and business + # quota, independently of the blocked business Host. + # Plans, dry runs and recovery of an original journal never poll. + if "external_evidence_poll" in normalize_required_capabilities(args.available_capabilities): + from ..capabilities.issue_fix.pr_wait_reconcile import reconcile_pr_wait_dependencies + + observation_started = time.monotonic_ns() + try: + pr_dependency_observation = reconcile_pr_wait_dependencies( + registry_path=registry_path, runtime_root_arg=str(runtime_root), + goal_id=args.goal_id, agent_id=args.agent_id, + ) + pr_dependency_observation["state_written"] = bool( + pr_dependency_observation["external_read_count"] + ) + except Exception as exc: # noqa: BLE001 - optional observation is fail-open + # A lost provider/log response is not proof of zero reads + # or zero local effects. Preserve that uncertainty while + # the ordinary decision owner still checks all admission. + pr_dependency_observation = { + "ok": False, "degraded": True, "failure_count": 1, + "error_category": type(exc).__name__, + "external_read_count": None, "state_written": None, + } + pr_dependency_observation["duration_ms"] = ( + time.monotonic_ns() - observation_started + ) // 1_000_000 from ..capabilities.semantic_preference.agent_preferences import extend_turn_start_dispatch as extend_preferences turn_start_hook_dispatch = extend_preferences(turn_start_hook_dispatch, runtime_root=runtime_root, registry_path=registry_path, goal_id=args.goal_id, agent_id=args.agent_id) @@ -325,6 +370,7 @@ def handle_turn_command( # never run its command as an arbitrary host/shell adapter. payload.update(mode="run_once", status="capability_action_required", execute=bool(args.execute), executed=False) + _attach_pr_dependency_observation(payload, pr_dependency_observation) print_payload(payload, output_format(args), _render_loopx_turn_plan_markdown) return 0 @@ -351,6 +397,7 @@ def handle_turn_command( payload = build_turn_error_payload( payload, exc, turn_command=args.turn_command, ) + _attach_pr_dependency_observation(payload, pr_dependency_observation) renderer = ( _render_loopx_turn_execution_markdown if args.turn_command == "run-once" diff --git a/loopx/cli_commands/turn_rendering.py b/loopx/cli_commands/turn_rendering.py index 136deda1f7..ac3ff4ef49 100644 --- a/loopx/cli_commands/turn_rendering.py +++ b/loopx/cli_commands/turn_rendering.py @@ -13,6 +13,33 @@ ) +def _pr_dependency_observation_lines(payload: Mapping[str, object]) -> list[str]: + observation = payload.get("pr_dependency_observation") + if not isinstance(observation, dict): + return [] + lines = [ + f"- PR dependency reads: {observation.get('external_read_count')}", + f"- PR observation duration_ms: {observation.get('duration_ms')}", + f"- PR dependency observation degraded: {observation.get('degraded')}", + f"- PR unavailable dependencies: {observation.get('unavailable_dependency_count')}", + f"- PR observation state_written: {observation.get('state_written')}", + ] + if observation.get("error_category"): + lines.append(f"- PR observation error: {observation['error_category']}") + for result in observation.get("results", []): + lines.append(f"- PR dependency {result['pr_ref']}: {result['state']}" + + (f" ({result['error_category']})" if result.get("error_category") else "")) + plan = observation.get("plan") + if isinstance(plan, dict): + lines.append(f"- PR retry interval_seconds: {plan.get('poll_interval_seconds')}") + observed = {result["pr_ref"] for result in observation.get("results", [])} + retained_failures = [target for target in plan.get("waiting_targets", []) + if target.get("last_observed_state") == "FAILED" and target["pr_ref"] not in observed] + for target in retained_failures[:4]: + lines.append(f"- PR dependency {target['pr_ref']}: FAILED; next_check_at: {target['next_due_at']}") + return lines + + def build_turn_error_payload( planned: dict[str, Any], exc: Exception, *, turn_command: str, execution_started: bool = False, @@ -72,6 +99,7 @@ def render_loopx_turn_plan_markdown(payload: dict[str, object]) -> str: capability = payload.get("capability_action") if isinstance(payload.get("capability_action"), dict) else {} intent = capability.get("intent") if isinstance(capability.get("intent"), dict) else {} envelope = payload.get("turn_envelope") + observation = payload.get("pr_dependency_observation") return "\n".join( [ "# LoopX Turn Plan", @@ -79,7 +107,10 @@ def render_loopx_turn_plan_markdown(payload: dict[str, object]) -> str: f"- execution_mode: {host.get('execution_mode')}", f"- route: {route.get('kind')}", f"- would_invoke_host: {route.get('would_invoke_host')}", - "- side_effects: none", + ("- side_effects: PR dependency observation" + if isinstance(observation, dict) and observation.get("state_written") is not False + else "- side_effects: none"), + *_pr_dependency_observation_lines(payload), *(["- capability_action: required (not executed)", f"- next_command: {capability.get('command') or intent.get('command')}"] if capability else []), *turn_envelope_budget_warning_lines( @@ -164,6 +195,7 @@ def render_loopx_turn_execution_markdown(payload: dict[str, object]) -> str: f"- host_invoked: {effects.get('host_invoked')}", f"- state_written: {effects.get('state_written')}", f"- quota_spent: {effects.get('quota_spent')}", + *_pr_dependency_observation_lines(payload), *([f"- error: {payload['error']}"] if payload.get("error") else []), *([f"- journal_readback: {journal_observation.get('status')}", f"- recorded_effects: {journal_observation.get('recorded_effects')}", diff --git a/tests/capabilities/test_pr_wait_observation.py b/tests/capabilities/test_pr_wait_observation.py index 3fa5a6ee4b..56963c7449 100644 --- a/tests/capabilities/test_pr_wait_observation.py +++ b/tests/capabilities/test_pr_wait_observation.py @@ -1,6 +1,11 @@ """A closed queue entry must still resume its exact canonical PR dependency.""" from __future__ import annotations +import contextlib +import io +import json +import sys + import pytest from subprocess import TimeoutExpired @@ -11,6 +16,371 @@ ) +def _native_wait_turn(tmp_path, monkeypatch, provider, *, claimed_by="agent-a"): + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, runtime, _ = promoted_create_fixture( + tmp_path, provider=provider, + registered_agents=["agent-a", claimed_by] if claimed_by != "agent-a" else None, + ) + project = tmp_path / "project" + data = json.loads(registry.read_text()) + data["goals"][0]["quota"] = {"compute": 0} + registry.write_text(json.dumps(data)) + todo_id = add_goal_todo( + registry_path=registry, goal_id="goal-a", role="agent", + text="Continue after the exact dependency merges", claimed_by=claimed_by, + resume_when="pr_merged:owner/repo#42", + )["todo_id"] + argv = [ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "goal-a", "--agent-id", "agent-a", + "--host", "generic-cli", "--project", str(project), "--scan-root", str(project), + "--host-adapter-command-json", json.dumps([ + sys.executable, "-c", "from pathlib import Path; Path('host-started').touch(); " + "raise AssertionError('business Host must remain paused')", + ]), "--available-capability", "external_evidence_poll", "--no-global-sync", "--execute", + ] + return registry, runtime, project, todo_id, argv + + +def _invoke_turn(argv): + from loopx.cli import main + + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = main(argv) + return code, json.loads(output.getvalue()) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_executing_turn_observes_dependency_before_business_quota( + tmp_path, monkeypatch, provider, +): + from loopx.capabilities.issue_fix import pr_lifecycle + from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path + + registry, runtime, project, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + before = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] + calls = [] + + def fetch(reference, **kwargs): + calls.append(reference) + return {"state": "MERGED", "mergedAt": "2026-10-01T00:00:00Z", + "url": "https://github.com/owner/repo/pull/42"} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + code, result = _invoke_turn([*argv, "--turn-instance-id", "observe-before-business-1"]) + assert code == 1, result + assert result["error"] == "LoopX Turn route is not host executable", result + after = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] + assert after["resume_ready"] is True, result + assert result["pr_dependency_observation"]["merged_count"] == 1 + assert result["pr_dependency_observation"]["state_written"] is True + assert not (project / "host-started").exists() + assert len(calls) == 1 + for key in ("text", "status", "claimed_by", "resume_when", "required_write_scopes"): + assert after.get(key) == before.get(key) + events = load_rollout_events(rollout_event_log_path(runtime, "goal-a")) + assert sum(event["event_kind"] == "pr_merge" for event in events) == 1 + assert not any(event["event_kind"] in {"quota_spend", "todo_complete"} for event in events) + + code, replay = _invoke_turn([*argv, "--turn-instance-id", "observe-before-business-2"]) + assert code == 1, replay + assert replay["error"] == "LoopX Turn route is not host executable", replay + assert len(calls) == 1 + assert replay["pr_dependency_observation"]["external_read_count"] == 0 + assert replay["pr_dependency_observation"]["state_written"] is False + assert not (project / "host-started").exists() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_turn_observes_dependencies_synchronized_during_turn_start(tmp_path, monkeypatch, provider): + from loopx.capabilities.issue_fix import pr_lifecycle + from loopx.cli_commands import lark_inbox + + registry, _, project, _, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + synchronized = [] + + def sync_inbox(**kwargs): + synchronized.append(add_goal_todo( + registry_path=registry, goal_id="goal-a", role="agent", claimed_by="agent-a", + text="Continue the synchronized dependency", resume_when="pr_merged:owner/repo#43", + )["todo_id"]) + return {} + + calls = [] + + def fetch(reference, **kwargs): + calls.append(reference["number"]) + return {"state": "MERGED", "mergedAt": "2026-10-01T00:00:00Z", + "url": f"https://github.com/owner/repo/pull/{reference['number']}"} + + monkeypatch.setattr(lark_inbox, "dispatch_goal_lark_turn_start_hooks", sync_inbox) + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + _, result = _invoke_turn(argv) + assert set(calls) == {42, 43}, result + assert list_goal_todos( + registry_path=registry, goal_id="goal-a", todo_id=synchronized[0], + )["todo"]["resume_ready"] is True + assert not (project / "host-started").exists() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("mode", ["absent-capability", "dry-run", "plan"]) +def test_read_only_and_unconfigured_turns_do_not_observe(tmp_path, monkeypatch, provider, mode): + from loopx.capabilities.issue_fix import pr_lifecycle + from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path + + registry, runtime, project, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + calls = [] + + def fetch(*args, **kwargs): + calls.append(args) + raise AssertionError("this invocation must not perform an external observation") + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + if mode == "absent-capability": + index = argv.index("--available-capability") + del argv[index:index + 2] + elif mode == "dry-run": + argv.remove("--execute") + else: + argv = [*argv[:argv.index("turn")], "turn", "plan", "--goal-id", "goal-a", + "--agent-id", "agent-a", "--host", "generic-cli", "--scan-root", str(project), + "--available-capability", "external_evidence_poll"] + _, result = _invoke_turn(argv) + assert "pr_dependency_observation" not in result + assert calls == [] + assert list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"]["resume_ready"] is False + assert not (project / "host-started").exists() + events = load_rollout_events(rollout_event_log_path(runtime, "goal-a")) + assert not any(event["event_kind"] in {"pr_merge", "quota_spend", "todo_complete"} + or event.get("classification") == "pr_wait_observation" for event in events) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("outcome", ["open", "closed", "timeout", "unauthenticated", "wrong-repo", "missing-timestamp"]) +def test_turn_observation_failure_and_retry_are_readable_without_business_effects( + tmp_path, monkeypatch, provider, outcome, +): + from loopx.capabilities.issue_fix import pr_lifecycle + from loopx.cli_commands.turn_rendering import render_loopx_turn_execution_markdown + from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path + + registry, runtime, project, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + calls = [] + + def fetch(*args, **kwargs): + calls.append(args) + if outcome == "timeout": + raise TimeoutExpired("gh", 10) + if outcome == "unauthenticated": + raise RuntimeError("metadata provider unavailable") + return {"url": "https://github.com/other/repo/pull/42" if outcome == "wrong-repo" + else "https://github.com/owner/repo/pull/42", + "state": "OPEN" if outcome == "open" else "CLOSED" if outcome == "closed" else "MERGED", + "mergedAt": "2026-10-01T00:00:00Z" if outcome == "wrong-repo" else None} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + failed = outcome not in {"open", "closed"} + for number in range(2): + _, result = _invoke_turn([*argv, "--turn-instance-id", f"pending-observation-{number}"]) + observation = result["pr_dependency_observation"] + assert observation["degraded"] is failed + assert observation["external_read_count"] == (1 if number == 0 else 0) + assert observation["unavailable_dependency_count"] == (1 if failed else 0) + assert observation["state_written"] is (number == 0) + assert f"PR dependency observation degraded: {failed}" in render_loopx_turn_execution_markdown(result) + assert len(calls) == 1 + after = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] + assert after["resume_ready"] is False + assert after["resume_when"] == "pr_merged:owner/repo#42" + assert after["status"] == "open" and after["claimed_by"] == "agent-a" + assert not (project / "host-started").exists() + events = load_rollout_events(rollout_event_log_path(runtime, "goal-a")) + assert not any(event["event_kind"] in {"pr_merge", "quota_spend", "todo_complete"} for event in events) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_partial_observation_write_remains_unknown_and_does_not_run_business( + tmp_path, monkeypatch, provider, +): + from loopx.capabilities.issue_fix import pr_lifecycle, pr_wait_reconcile + + registry, _, project, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", lambda *args, **kwargs: { + "url": "https://github.com/owner/repo/pull/42", "state": "MERGED", "mergedAt": "2026-10-01T00:00:00Z", + }) + + def lose_observation_receipt(*args, **kwargs): + raise OSError("local observation receipt unavailable") + + monkeypatch.setattr(pr_wait_reconcile, "append_rollout_event", lose_observation_receipt) + _, result = _invoke_turn(argv) + observation = result["pr_dependency_observation"] + assert observation["ok"] is False and observation["degraded"] is True + assert observation["external_read_count"] is None + assert observation["state_written"] is None + assert observation["error_category"] == "OSError" + assert list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"]["resume_ready"] is True + assert not (project / "host-started").exists() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_fresh_turns_reconcile_a_later_merge_on_the_retained_cadence(tmp_path, monkeypatch, provider): + from loopx.capabilities.issue_fix import pr_lifecycle, pr_wait_reconcile + from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path + + registry, runtime, project, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + clock = ["2026-10-01T01:00:00Z"] + monkeypatch.setattr(pr_wait_reconcile, "now_utc_iso", lambda: clock[0]) + calls = [] + + def fetch(*args, **kwargs): + calls.append(args) + return {"url": "https://github.com/owner/repo/pull/42", + "state": "OPEN" if len(calls) == 1 else "MERGED", + "mergedAt": "2026-10-01T01:10:00Z" if len(calls) > 1 else None} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + for number, (at, count, ready) in enumerate([ + ("2026-10-01T01:00:00Z", 1, False), ("2026-10-01T01:29:59Z", 1, False), + ("2026-10-01T01:30:00Z", 2, True), ("2026-10-01T02:30:00Z", 2, True), + ]): + clock[0] = at + _invoke_turn([*argv, "--turn-instance-id", f"native-cadence-{number}"]) + after = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] + assert after["resume_ready"] is ready + assert len(calls) == count + events = load_rollout_events(rollout_event_log_path(runtime, "goal-a")) + merges = [event for event in events if event["event_kind"] == "pr_merge"] + assert len(merges) == 1 and merges[0]["recorded_at"] == "2026-10-01T01:10:00Z" + assert not any(event["event_kind"] in {"quota_spend", "todo_complete"} for event in events) + assert not (project / "host-started").exists() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_turn_observer_does_not_take_another_agents_dependency(tmp_path, monkeypatch, provider): + from loopx.capabilities.issue_fix import pr_lifecycle + + registry, _, project, todo_id, argv = _native_wait_turn( + tmp_path, monkeypatch, provider, claimed_by="agent-b", + ) + calls = [] + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", lambda *args, **kwargs: calls.append(args)) + _, result = _invoke_turn(argv) + assert result["pr_dependency_observation"]["external_read_count"] == 0 + after = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] + assert after["claimed_by"] == "agent-b" and after["resume_ready"] is False + assert calls == [] and not (project / "host-started").exists() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_observation_does_not_resume_a_stopped_goals_business(tmp_path, monkeypatch, provider): + from loopx.capabilities.issue_fix import pr_lifecycle + from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path + + registry, runtime, project, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + data = json.loads(registry.read_text()) + data["goals"][0]["quota"] = {"compute": 1} + data["goals"][0]["activation"] = { + "schema_version": "loopx_goal_activation_v1", "state": "stopped", + "updated_at": "2026-10-01T00:00:00Z", "reason": "Stopped by owner", "actor_kind": "owner", + } + registry.write_text(json.dumps(data)) + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", lambda *args, **kwargs: { + "url": "https://github.com/owner/repo/pull/42", "state": "MERGED", "mergedAt": "2026-10-01T00:00:00Z", + }) + _, result = _invoke_turn(argv) + assert result["pr_dependency_observation"]["merged_count"] == 1 + assert list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"]["resume_ready"] is True + assert json.loads(registry.read_text())["goals"][0]["activation"]["state"] == "stopped" + assert not (project / "host-started").exists() + events = load_rollout_events(rollout_event_log_path(runtime, "goal-a")) + assert not any(event["event_kind"] in {"quota_spend", "todo_complete"} for event in events) + + +def test_original_journal_recovery_does_not_run_fresh_observation(tmp_path, monkeypatch): + from loopx.capabilities.issue_fix import pr_lifecycle + from loopx.cli_commands import turn, turn_run_once + + _, _, _, _, argv = _native_wait_turn(tmp_path, monkeypatch, "file") + calls = [] + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", lambda *args, **kwargs: calls.append(args)) + monkeypatch.setattr(turn, "load_loopx_turn_plan_from_journal", lambda *args, **kwargs: { + "turn_envelope": {"agent_id": "agent-a"}, + }) + monkeypatch.setattr(turn_run_once, "execute_turn_run_once", lambda *args, **kwargs: { + "ok": True, "status": "committed", "effects": { + "host_invoked": False, "state_written": False, "quota_spent": False, + }, + }) + # This isolates the existing recovery-entry wiring; real journal replay is + # covered separately by the native Turn driver suite. + code, result = _invoke_turn([*argv, "--resume-turn-key", "sha256:" + "a" * 64]) + assert code == 0, result + assert calls == [] and "pr_dependency_observation" not in result + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_native_turn_adopts_merge_then_replays_original_journal_once(tmp_path, monkeypatch, provider): + from loopx.capabilities.issue_fix import pr_lifecycle + from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path + from tests.test_loopx_turn_driver import _write_live_fixture + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, provider=provider, todo_metadata_extra="resume_when=pr_merged:owner/repo#42", + ) + calls = [] + + def fetch(*args, **kwargs): + calls.append(args) + return {"url": "https://github.com/owner/repo/pull/42", "state": "MERGED", + "mergedAt": "2026-10-01T00:00:00Z"} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + runner = tmp_path / "protocol_host.py" + runner.write_text( + "import json\nfrom pathlib import Path\n" + "def run_dsh_turn(**kwargs):\n" + " root = Path(kwargs['workspace'])\n" + " counter = root / 'host-count'\n" + " counter.write_text(str(int(counter.read_text()) + 1 if counter.exists() else 1))\n" + " (root / 'validated-artifact').write_text('continued')\n" + " return json.dumps({'result_kind':'validated_progress'," + "'classification':'dependency_followthrough','summary':'Validated the resumed fixture work.'," + "'delivery_batch_scale':'implementation','delivery_outcome':'outcome_progress'," + "'delivery_boundary':'in_flight_continuation','recommended_action':'Continue the original task.'})\n" + ) + argv = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--host", "dsh", "--dsh-runner", str(runner), "--project", str(project), + "--scan-root", str(project), "--available-capability", "external_evidence_poll", + "--validation-command-json", json.dumps([sys.executable, "-c", + "from pathlib import Path; assert Path('validated-artifact').read_text() == 'continued'"]), + "--no-global-sync", "--execute"] + before = list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture", todo_id="todo_fixture0001")["todo"] + assert before["resume_ready"] is False + code, result = _invoke_turn([*argv, "--turn-instance-id", "adopt-observed-merge"]) + assert code == 0 and result["status"] == "committed", result + assert (project / "host-count").read_text() == "1" + observation = result["pr_dependency_observation"] + assert observation["merged_count"] == 1 and observation["duration_ms"] >= 0 + after = list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture", todo_id="todo_fixture0001")["todo"] + assert after["resume_ready"] is True + for key in ("text", "status", "claimed_by", "resume_when"): + assert after[key] == before[key] + + code, replay = _invoke_turn([*argv, "--resume-turn-key", result["resume_turn_key"]]) + assert code == 0 and replay["status"] == "committed", replay + assert "pr_dependency_observation" not in replay + assert len(calls) == 1 and (project / "host-count").read_text() == "1" + events = load_rollout_events(rollout_event_log_path(runtime, "loopx-turn-fixture")) + assert sum(event["event_kind"] == "pr_merge" for event in events) == 1 + assert sum(event["event_kind"] == "quota_spend" for event in events) == 1 + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_prequota_observes_merged_dependency_without_open_queue_membership(tmp_path, monkeypatch, provider): from loopx.capabilities.issue_fix import pr_lifecycle From 832e1d7eabbfdba1b99392e2f5902b1edb8e8bd8 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 16:00:38 +0800 Subject: [PATCH 2/4] fix(turn): retain selected dependency proof through signed host input Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/agent-loop-effect-interpreter-v0.md | 9 + .../agent-loop-effect-interpreter-v0.zh-CN.md | 5 + docs/reference/todo-continuation-readback.md | 17 ++ .../quota/selected_todo_projection.py | 3 + loopx/control_plane/quota/turn_envelope.ts | 1 + .../test_pr_wait_consumer_adoption.py | 162 ++++++++++++++++++ tests/control_plane_ts/turn_envelope.test.ts | 27 +++ 7 files changed, 224 insertions(+) create mode 100644 tests/control_plane/test_pr_wait_consumer_adoption.py diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md index 8b5724aa3b..b69612ad67 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md @@ -484,6 +484,15 @@ This is a native CLI adoption slice; installed heartbeat declarations, App/Lark caller adoption, sustained live scheduling, model outcomes and the three convergence acceptances remain open. +The same captured decision now retains the selected Task's existing continuation +condition and readiness through the full selection projection, signed envelope +and actual protocol-host input. File/SQLite qualification distinguishes two +repositories' identically numbered PRs, preserves the original event time and +long requirement tail, validates subsequent work, and rejects proof tampering. +Read-only managed-step and original-journal replay add no observation or debit. +This closes that fact-delivery omission; installed adoption and model outcomes +remain outside this qualification. + Bounded planning context now remains inline instead of forcing a full-decision read under envelope budget pressure. Its existing owner retains the 3 KiB bound; the shared DSH/cloud adapter transports the signed contribution and, for saved diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md index fd2b0dd3e8..483ee4b4e8 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md @@ -337,6 +337,11 @@ envelope 投影,Codex CLI 保持关闭隔离;这不代表安装态 heartbeat 本切片只完成原生 CLI 接入;安装态 heartbeat 声明、App/Lark 调用方采用、持续真实调度、 模型结果与以下三批收敛验收保持开放。 +同次捕获决策中的所选 Task 继续条件与满足状态,现在沿完整选择投影、签名 envelope +到实际协议宿主输入完整保留。File/SQLite 验证区分不同仓库的同号 PR,保留原事件时间 +和长要求尾部,独立验证后续工作并拒绝证明篡改。只读 managed-step 与原 journal +重放不新增观察或扣费。本批补齐事实交付遗漏,安装采用与模型结果仍另行验收。 + 有界规划上下文现在保持内联,不再因短包预算压力而强制读取整份决策;既有 owner 的 3 KiB 限制保留。共享 DSH/cloud adapter 传递签名内容,历史引用同时保留原捕获路由和 哈希。thin renderer 仅在存在 `agent_context.detail_ref` 时要求能力补读,不能把也用于 diff --git a/docs/reference/todo-continuation-readback.md b/docs/reference/todo-continuation-readback.md index 9207e1e710..546b099cc6 100644 --- a/docs/reference/todo-continuation-readback.md +++ b/docs/reference/todo-continuation-readback.md @@ -56,6 +56,16 @@ actual network latency and total host costs remain workload-dependent. This qualifies the native CLI bridge. Installed heartbeat declarations, App/Lark caller adoption and sustained live scheduling remain separate boundaries. +Full decisions retain the selected Task's existing `resume_when`, `resume_ready` +and complete `resume_condition`; TurnEnvelope carries the same fields in +`action.selected_todo` and signs them with the action. Native Host inputs retain +that proof, including its repository, event identity and original event time, +alongside the complete Task requirements. This is observation transport; +execution still needs the ordinary current admission. The projection does not +recompute satisfaction or substitute another Task's proof. Tasks without these +fields keep their prior shape, and saved journal envelopes keep their original +signed contents. `managed-step` remains a read-only controller and does not poll. + `pr_merged:owner/repo#42` 只标识依赖;缺少合并事件不证明 GitHub 仍在等待。 历史交付监督不再豁免这种未观察等待,因果阻塞结算仍要求已注册的 `monitor_changed` 或 `todo_done` 目标及当前回读。 @@ -87,6 +97,13 @@ claim、lease 与宿主启动仍由原准入约束。plan、dry run 与 `--resum 仍依赖工作负载。本切片验证原生 CLI 接入;安装态 heartbeat 声明、App/Lark 调用方采用 与持续真实调度仍分别验收。 +完整决策保留所选 Task 既有的 `resume_when`、`resume_ready` 与完整 +`resume_condition`;TurnEnvelope 将相同字段置于 `action.selected_todo`,纳入 +动作签名。原生 Host 输入同时保留该证明的仓库、事件身份、原事件时间与完整任务要求。 +这些是观察事实的传输,执行仍需正常当前准入;投影不重判满足,也不借用另一 Task 的 +证明。缺少这些字段的任务保持原形态,已保存 journal 保留原签名内容。 +`managed-step` 仍是只读控制器,不执行观察轮询。 + ## Completion dependency cycles / 完成依赖环 An unfinished `todo_done` chain that reaches a cycle now projects diff --git a/loopx/control_plane/quota/selected_todo_projection.py b/loopx/control_plane/quota/selected_todo_projection.py index eabb08fd72..54249077cb 100644 --- a/loopx/control_plane/quota/selected_todo_projection.py +++ b/loopx/control_plane/quota/selected_todo_projection.py @@ -27,6 +27,9 @@ "blocks_agent", "excluded_agents", "unblocks_todo_id", + "resume_when", + "resume_ready", + "resume_condition", "target_key", "next_due_at", "expires_at", diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 7757858da8..2521c04110 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -233,6 +233,7 @@ function selectedTodo(payload: JsonObject, recommendedAction: string | null): Js "task_repository", "continuation_policy", "claimed_by", "bound_agent", "goal_bound", "blocks_agent", "unblocks_todo_id", "next_due_at", "expires_at", "selected_by", "confidence", + "resume_when", "resume_ready", "resume_condition", ]) { if (source[field] !== null && source[field] !== undefined) compact[field] = source[field]; } diff --git a/tests/control_plane/test_pr_wait_consumer_adoption.py b/tests/control_plane/test_pr_wait_consumer_adoption.py new file mode 100644 index 0000000000..5ad3d26f71 --- /dev/null +++ b/tests/control_plane/test_pr_wait_consumer_adoption.py @@ -0,0 +1,162 @@ +"""The selected dependency's proof reaches the actual consumer and validated work.""" +from __future__ import annotations + +import copy +import json +import shlex +import sys +from pathlib import Path + +import pytest + +from loopx.capabilities.issue_fix import pr_lifecycle +from loopx.cli_commands import turn +from loopx.control_plane.quota.turn_envelope import build_turn_envelope +from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority +from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path +from loopx.todos import add_goal_todo, list_goal_todos +from tests.capabilities.test_pr_wait_observation import _invoke_turn +from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime +from tests.test_loopx_turn_driver import _write_live_fixture + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("observer_entry", ["native", "heartbeat"]) +def test_selected_merge_proof_survives_capture_and_actual_host_adoption( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, observer_entry: str, +) -> None: + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, provider=provider, todo_metadata_extra="resume_when=pr_merged:other/repo#42", + ) + requirement = "Preserve the accepted requirement. " * 22 + "Require the terminal independent artifact marker." + todo_id = add_goal_todo( + registry_path=registry, goal_id="loopx-turn-fixture", role="agent", claimed_by="codex-fixture", + text=requirement, priority="P0", action_kind="fixture", resume_when="pr_merged:owner/repo#42", + )["todo_id"] + before = list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture", todo_id=todo_id)["todo"] + assert before["resume_ready"] is False + calls = [] + + def fetch(reference, **kwargs): + calls.append(reference) + return {"url": f"https://github.com/{reference['repo']}/pull/{reference['number']}", "state": "MERGED", + "mergedAt": "2026-10-01T00:00:00Z" if reference["repo"] == "owner/repo" + else "2026-10-02T00:00:00Z"} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] + if observer_entry == "heartbeat": + prompts = [] + for mode in ("full", "thin"): + code, prompt = _invoke_turn([ + *prefix, "heartbeat-prompt", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--codex-app", "--available-capability", "external_evidence_poll", f"--{mode}", + ]) + assert code == 0, prompt + prompts.append(prompt) + command = prompts[0]["pr_review_pre_quota_command"] + for number, prompt in enumerate(prompts): + # Execute the command actually carried by each generated bootstrap. + # Only output format is selected by the harness; its route is intact. + generated = next(line for line in prompt["task_body"].splitlines() if line.strip() == command) + code, observation = _invoke_turn(["--format", "json", *shlex.split(generated)[1:]]) + assert code == 0, observation + assert observation["checks"]["pr_dependencies"]["external_read_count"] == (2 if number == 0 else 0) + + captured = [] + project_envelope = turn.build_turn_envelope + + def capture(decision, **kwargs): + captured.append(copy.deepcopy(decision)) + return project_envelope(decision, **kwargs) + + monkeypatch.setattr(turn, "build_turn_envelope", capture) + runner = tmp_path / "consumer.py" + runner.write_text( + "import json\nfrom pathlib import Path\n" + "def run_dsh_turn(**kwargs):\n" + " root = Path(kwargs['workspace'])\n" + " prompt = kwargs['prompt']\n" + " (root / 'consumer-input.txt').write_text(prompt)\n" + " authority = json.JSONDecoder().raw_decode(prompt.split('Turn authority JSON:\\n', 1)[1])[0]\n" + " selected = authority['selected_todo']\n" + f" assert selected['todo_id'] == {todo_id!r}\n" + " assert selected['resume_when'] == 'pr_merged:owner/repo#42'\n" + " assert selected['resume_ready'] is True\n" + " condition = selected['resume_condition']\n" + " assert condition['satisfied'] is True\n" + " assert condition['matched_pr_ref'] == 'owner/repo#42'\n" + " assert condition['matched_event_at'] == '2026-10-01T00:00:00Z'\n" + " task = next(s['content']['todo'] for s in authority['work_context']['sources']\n" + " if s.get('source') == 'selected_todo')\n" + f" assert task['text'] == {before['text']!r}\n" + " counter = root / 'consumer-count'\n" + " counter.write_text(str(int(counter.read_text()) + 1 if counter.exists() else 1))\n" + " (root / 'validated-artifact').write_text('terminal independent artifact marker')\n" + " return json.dumps({'result_kind':'validated_progress','classification':'dependency_followthrough'," + "'summary':'Validated the original dependency-bound fixture work.','delivery_batch_scale':'implementation'," + "'delivery_outcome':'outcome_progress','delivery_boundary':'in_flight_continuation'," + "'recommended_action':'Continue the original accepted work.'})\n" + ) + argv = [*prefix, "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--host", "dsh", "--dsh-runner", str(runner), "--project", str(project), "--scan-root", str(project), + "--todo-id", todo_id, "--turn-instance-id", "consumer-adoption", "--available-capability", "external_evidence_poll", + "--validation-command-json", json.dumps([sys.executable, "-c", "from pathlib import Path; " + "assert Path('validated-artifact').read_text() == 'terminal independent artifact marker'"]), + "--no-global-sync", "--execute"] + code, result = _invoke_turn(argv) + assert code == 0 and result["status"] == "committed", json.dumps(result, indent=2) + assert result["pr_dependency_observation"]["external_read_count"] == (2 if observer_entry == "native" else 0) + assert len(captured) == 1 + full = captured[0] + envelope = build_turn_envelope(full) + expected = list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture", todo_id=todo_id)["todo"] + assert expected["resume_ready"] is True + for field in ("resume_when", "resume_ready", "resume_condition"): + assert full["selected_todo"][field] == expected[field] + assert envelope["action"]["selected_todo"][field] == expected[field] + assert expected["resume_condition"]["matched_pr_ref"] == "owner/repo#42" + assert expected["resume_condition"]["matched_event_at"] == "2026-10-01T00:00:00Z" + assert envelope["action_signature"]["matches"] is True + prompt = (project / "consumer-input.txt").read_text() + consumed = json.JSONDecoder().raw_decode(prompt.split("Turn authority JSON:\n", 1)[1])[0] + assert consumed["selected_todo"]["resume_condition"] == expected["resume_condition"] + assert "Require the terminal independent artifact marker." in prompt + + # A matching signature must cover readiness and the exact event provenance. + for field, value in (("resume_ready", False), ("resume_condition", {"satisfied": True, "matched_pr_ref": "other/repo#42"})): + changed = copy.deepcopy(envelope) + changed["action"]["selected_todo"][field] = value + with pytest.raises(ValueError, match="action signature"): + extract_turn_authority({"turn_envelope": changed}) + + for field in ("text", "status", "claimed_by", "resume_when"): + assert expected[field] == before[field] + # A new pending dependency makes accidental polling by either read/replay visible. + pending = add_goal_todo( + registry_path=registry, goal_id="loopx-turn-fixture", role="agent", claimed_by="codex-fixture", + text="Continue a separately observed dependency", resume_when="pr_merged:owner/repo#43", + )["todo_id"] + events = load_rollout_events(rollout_event_log_path(runtime, "loopx-turn-fixture")) + assert len(calls) == 2 and {call["repo"] for call in calls} == {"owner/repo", "other/repo"} + assert sum(event["event_kind"] == "quota_spend" for event in events) == 1 + assert sum(event["event_kind"] == "pr_merge" for event in events) == 2 + assert not any(event["event_kind"] == "todo_complete" for event in events) + + # Managed-step remains a read-only controller even with polling declared. + revision = list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture")["authority_read"]["provider_revision"] + code, managed = _invoke_turn([*prefix, "turn", "managed-step", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--host", "dsh", "--turn-key", result["resume_turn_key"], + "--scan-root", str(project), "--available-capability", "external_evidence_poll"]) + assert code == 1 and managed["error"] == "managed step requires a failed Turn journal", managed + assert list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture")["authority_read"]["provider_revision"] == revision + replay_argv = list(argv) + for fresh_option in ("--todo-id", "--turn-instance-id"): + selection_index = replay_argv.index(fresh_option) + del replay_argv[selection_index:selection_index + 2] + code, replay = _invoke_turn([*replay_argv, "--resume-turn-key", result["resume_turn_key"]]) + assert code == 0 and replay["status"] == "committed", replay + assert len(calls) == 2 and (project / "consumer-count").read_text() == "1" + assert list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture", todo_id=pending)["todo"]["resume_ready"] is False + assert load_rollout_events(rollout_event_log_path(runtime, "loopx-turn-fixture")) == events diff --git a/tests/control_plane_ts/turn_envelope.test.ts b/tests/control_plane_ts/turn_envelope.test.ts index 06cced35b3..e224b0f9eb 100644 --- a/tests/control_plane_ts/turn_envelope.test.ts +++ b/tests/control_plane_ts/turn_envelope.test.ts @@ -830,6 +830,33 @@ test("selected work aliases only an exact complete recommendation", () => { assert.equal(selected.text, undefined); }); +test("selected dependency keeps its original readiness and event proof signed", () => { + const source = payload(); + const original = source.selected_todo as Record; + Object.assign(original, { + resume_when: "pr_merged:owner/repo#42", resume_ready: true, + resume_condition: {schema_version: "todo_resume_condition_v0", kind: "pr_merged", + satisfied: true, matched_pr_ref: "owner/repo#42", matched_event_id: "original-merge", + matched_event_at: "2026-10-01T00:00:00Z", source: "rollout_event_log"}, + }); + const envelope = buildTurnEnvelope({payload: source, protocol_action_fields: protocolActionFields, + scheduler_execution_args: ""}); + const selected = (envelope.action as Record).selected_todo as Record; + for (const field of ["resume_when", "resume_ready", "resume_condition"]) { + assert.deepEqual(selected[field], original[field]); + } + const signed = structuredClone(turnEnvelopeActionSignatureDocument(envelope)); + selected.resume_ready = false; + assert.notDeepEqual(turnEnvelopeActionSignatureDocument(envelope), signed); + selected.resume_ready = true; + (selected.resume_condition as Record).matched_event_id = "another-event"; + assert.notDeepEqual(turnEnvelopeActionSignatureDocument(envelope), signed); + const historical = buildTurnEnvelope({payload: payload(), protocol_action_fields: protocolActionFields, + scheduler_execution_args: ""}); + const historicalSelected = (historical.action as Record).selected_todo as Record; + for (const field of ["resume_when", "resume_ready", "resume_condition"]) assert.equal(field in historicalSelected, false); +}); + test("signature document fails closed to the canonical signed dimensions", () => { const envelope = buildTurnEnvelope({ payload: payload(), From 62238dda5c2837b75249d2d734b547e6b65a56d9 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 18:03:10 +0800 Subject: [PATCH 3/4] refactor(turn): share registered dependency observation across entrypoints Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/agent-loop-effect-interpreter-v0.md | 29 +- .../agent-loop-effect-interpreter-v0.zh-CN.md | 17 +- docs/reference/protocols/loopx-turn-v0.md | 27 ++ docs/reference/todo-continuation-readback.md | 124 ++++--- loopx/capabilities/issue_fix/README.md | 17 + loopx/capabilities/issue_fix/pr_wait_hook.py | 132 ++++++++ .../issue_fix/pr_wait_reconcile.py | 22 ++ loopx/cli_commands/quota.py | 13 +- loopx/cli_commands/turn.py | 68 +--- loopx/cli_commands/turn_rendering.py | 42 +-- .../control_plane/capabilities/turn_start.py | 57 ++++ loopx/control_plane/capability_hooks.py | 5 + loopx/control_plane/capability_hooks.ts | 25 +- loopx/heartbeat_prequota.py | 25 +- tests/capabilities/test_pr_wait_lifecycle.py | 317 ++++++++++++++++++ .../capabilities/test_pr_wait_observation.py | 37 +- .../test_pr_wait_consumer_adoption.py | 9 +- tests/control_plane/test_pr_wait_history.py | 3 +- .../test_turn_start_capability_hooks.py | 11 + .../control_plane_ts/capability_hooks.test.ts | 25 ++ 20 files changed, 816 insertions(+), 189 deletions(-) create mode 100644 loopx/capabilities/issue_fix/pr_wait_hook.py create mode 100644 loopx/control_plane/capabilities/turn_start.py create mode 100644 tests/capabilities/test_pr_wait_lifecycle.py diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md index b69612ad67..b3adbfccbe 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md @@ -471,18 +471,23 @@ shape merely because of its size. Optional memory participation now has one compact, signed envelope projection; the Codex CLI adapter preserves default-off isolation. This does not qualify installed heartbeat/App adoption or model value. -Fresh executing native Turns now compose the existing opt-in PR dependency -observer before their governing decision. An exact merged observation can -satisfy the original wait without asking the blocked business Host to select -that observation. The TypeScript plan, retained merge evidence and ordinary -execution admission remain the owners; the CLI reports observation cost and -failure/uncertain-write readback. File/SQLite qualification covers paused -business compute, subsequent validated protocol-host work and original-journal -replay, with the original requirement, ownership and merge timestamp retained. -Plans, dry runs, missing declarations and journal recovery add no observation. -This is a native CLI adoption slice; installed heartbeat declarations, -App/Lark caller adoption, sustained live scheduling, model outcomes and the -three convergence acceptances remain open. +PR dependency observation now uses the existing registered TurnStart lifecycle +shared by executing quota guards, native Turns and compatibility prequota. +Permission comes from the typed Goal/Agent capability owner, including remembered +host observations and revocation; callers need no repeated flag when permission +already exists. Capability-owned diagnostics preserve domain policy outside core +Turn code. The existing complete-source TypeScript plan, merge-fact owner and +normal execution admission remain authoritative. Goal-local single flight and +durable per-dependency cadence prevent duplicate concurrent/due reads. +File/SQLite qualification covers paused/stopped business, generated full/thin +heartbeat commands, cross-entrypoint observation, revocation, unknown effects, +unrelated validated work after provider failure and original-journal replay. +App/Lark's shared context backend consumes canonical readiness without polling. +Plans, read-only/status/context paths, missing permission and recovery add no +observation. Installed executor adoption, legitimate recovery of existing live +owners, sustained scheduling, model outcomes and the three convergence +acceptances remain open. Generic memory lifecycle integration is a separate +remaining boundary; post-writeback hooks retain their effect-free contract. The same captured decision now retains the selected Task's existing continuation condition and readiness through the full selection projection, signed envelope diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md index 483ee4b4e8..6f12e2d792 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md @@ -329,13 +329,16 @@ R5 短包投影也完整保留已有 CLI 结算计划,包括 effect identity envelope 投影,Codex CLI 保持关闭隔离;这不代表安装态 heartbeat/App 或模型收益 已验收。 -新的原生 Turn 执行入口在 governing decision 前组合既有 opt-in PR 依赖观察器, -精确合并事实可满足原等待,不要求被该等待阻挡的业务 Host 主动选择观察工作。目标计划、 -保留的合并证据与正常执行准入继续归 TS 及原 owner;CLI 读回观察成本、失败与不确定 -写入。File/SQLite 验证覆盖业务额度暂停、随后经过独立校验的协议宿主工作和原 journal -重放,保留原要求、归属及合并时间。plan、dry run、缺少声明与 journal 恢复不添加观察。 -本切片只完成原生 CLI 接入;安装态 heartbeat 声明、App/Lark 调用方采用、持续真实调度、 -模型结果与以下三批收敛验收保持开放。 +PR 依赖观察现复用既有注册 TurnStart 生命周期,由执行性 quota guard、原生 Turn +与兼容 prequota 共用。权限来自 typed Goal/Agent capability owner,包括已记住的 +宿主观察与撤销;已有授权时不必逐命令补 flag。能力自有诊断使 PR 业务规则留在 core +Turn 之外;完整源 TS 计划、合并事实 owner 与正常执行准入仍为原权威。Goal 内 +single-flight 与每项依赖的持久节奏避免并发/到期读取重复。File/SQLite 验证覆盖业务 +额度暂停/停止、full/thin heartbeat 生成命令、跨入口观察、撤销、不确定效果、provider +失败后的无关独立校验工作及原 journal 重放。App/Lark 共用上下文后端只消费 canonical +就绪状态,不轮询。plan、只读/status/上下文、缺少权限与恢复路径均不添加观察。 +安装态执行器采用、既有真实 owner 的合法恢复、持续调度、模型结果与以下三批收敛验收 +仍开放。通用 memory 生命周期接入是另一项未完成边界;post-writeback hook 继续无效果。 同次捕获决策中的所选 Task 继续条件与满足状态,现在沿完整选择投影、签名 envelope 到实际协议宿主输入完整保留。File/SQLite 验证区分不同仓库的同号 PR,保留原事件时间 diff --git a/docs/reference/protocols/loopx-turn-v0.md b/docs/reference/protocols/loopx-turn-v0.md index fe80dc7b4b..d78c827f34 100644 --- a/docs/reference/protocols/loopx-turn-v0.md +++ b/docs/reference/protocols/loopx-turn-v0.md @@ -416,6 +416,33 @@ One driver tick has exactly these ordered phases: The driver may stop after any phase. A stop must return a typed result and must not silently continue with a different execution mode. +Executing admission entrypoints compose registered TurnStart observations +before collecting the decision. The capability owns permission resolution, +provider IO and bounded readback; the shared dispatcher validates the existing +registration/result contract and isolates failures. Core Turn rendering consumes +`readback.summary_lines`, without knowing the capability's domain. The same +registration is used by executing quota guards and compatibility prequota. +Read-only builders, plans, managed-step and original-journal recovery do not +invoke new observations. An observer may establish a fact during a business +wait or compute pause, but it cannot grant host execution or spend authority. + +TurnStart results may carry bounded capability-owned `readback` without an +Agent message-read obligation. Failed/partial/unavailable results may report +unknown external reads or local effects as `null`; external writes remain an +explicit boolean. If invocation precedes a rejected/lost result, the dispatcher +retains `uncertain_local_effects` for declared local writes and callers refresh +their decision projection. These extensions do not make post-writeback hooks +effectful or authorize new provider writes. + +执行准入入口在决策收集前组合注册的 TurnStart 观察。能力负责权限、provider IO 与 +有界读回;共享 dispatcher 校验既有契约并隔离失败,核心 renderer 只消费通用 +`readback.summary_lines`。执行 quota guard 与兼容 prequota 复用该注册。只读 builder、 +plan、managed-step 和原 journal 恢复不添加观察。业务等待或额度暂停中的事实观察 +不授予宿主执行或扣费权限。失败/退化/不可用结果可用 `null` 表示外部读取或本地效果 +未知,远端写入仍须显式布尔声明。执行后响应丢失/被拒绝时,dispatcher 为已声明的 +本地写入保留 `uncertain_local_effects`,调用方刷新决策投影;post-writeback hook +继续无效果,不由此获得 provider 写入权限。 + ### One Executor Per Turn Lane A **Turn lane** is one agent working one goal. Phase 5 launches exactly one diff --git a/docs/reference/todo-continuation-readback.md b/docs/reference/todo-continuation-readback.md index 546b099cc6..d96cd7eebf 100644 --- a/docs/reference/todo-continuation-readback.md +++ b/docs/reference/todo-continuation-readback.md @@ -14,10 +14,12 @@ does not prove that GitHub is still pending. Delivery-history supervision no longer exempts that unobserved wait. Causal blocked closeout continues to require a registered `monitor_changed` or `todo_done` target with current readback. -The existing `heartbeat-prequota --goal-id GOAL --agent-id AGENT` observes exact +The registered `issue-fix` TurnStart hook observes exact unfinished PR dependencies from the complete Todo head, including deferred work. It polls at most four distinct repository-qualified dependencies per invocation, -with a durable 30-minute cadence for successful and failed attempts. A missing +with a durable 30-minute cadence per dependency for successful and failed attempts. +Executing callers share one Goal-local single flight; a concurrent caller reports +observation in progress without another provider request. A missing repository binding is reported for repair; it never guesses from PR numbers. An exact merged observation appends the existing native `pr_merge` fact so the original condition becomes satisfied. Resume readback uses the same complete @@ -31,30 +33,49 @@ and provider failures cannot satisfy it. This observation changes no Todo status requirements, claims, leases or Goal acceptance. Deferred ready work still needs its normal lifecycle transition. -Generated heartbeat prompts include prequota only when the host declares -`--available-capability external_evidence_poll`; filesystem/shell declarations -alone do not opt in. Omitting that declaration keeps external observation off. -Explicit prequota invocation authorizes compact metadata reads and local -observation writeback, never merge, review, remote mutation or quota spend. -Read back with `todo list --goal-id GOAL --todo-id TODO --format json`. - -Fresh `turn run-once --execute` invocations with that same explicit capability -now call the existing PR dependency observer before collecting the governing -decision. Observation can satisfy the wait while business compute is paused; -the ordinary stop, quota, claim, lease and launch checks still govern execution. -Planning, dry runs and `--resume-turn-key` recovery perform no new observation. -Omitting the capability retains the previous path. Disabling the declaration -stops future automatic reads and retains previously verified merge facts. - -The native CLI returns `pr_dependency_observation` in JSON: remote read count, -observer duration in milliseconds, degraded/failure readback, the existing -target/cadence plan and observation `state_written`. Markdown summarizes those -counts, cadence and failures. Observation `state_written` is `null` after an uncertain provider/log failure; a failed response must -not claim zero local effects. Observation has no separate model call or business -quota debit. The existing four-target bound and 30-minute retry cadence apply; -actual network latency and total host costs remain workload-dependent. This -qualifies the native CLI bridge. Installed heartbeat declarations, App/Lark -caller adoption and sustained live scheduling remain separate boundaries. +The existing typed capability owner resolves `external_evidence_poll` from the +current Goal declaration, remembered host observations and invocation facts; +remembered revocation defeats inherited permission. No per-command flag is +required when the owner already supplies permission. Shell/filesystem access, +Todo requirements and explicit prequota invocation do not grant network access. +Unavailable permission sources return a failure without contacting GitHub. +Remove all inherited grants or record the capability as unavailable to stop +future automatic reads; previously verified merge facts remain usable. + +Fresh executing `quota should-run` and `turn run-once --execute` compose the +same registered hook before collecting the governing decision. Compatibility +`heartbeat-prequota --goal-id GOAL --agent-id AGENT` uses that hook too. Generated +full/thin heartbeat prompts retain their optional prequota command when the +host explicitly declares polling; their executing guard also resolves current +owner permission. Observation can continue while business compute is paused or +business execution is stopped, if observation remains authorized. The ordinary +stop, quota, claim, lease and launch checks still govern subsequent work. +Status, list, page/context reads, prompt generation, plans, dry runs, +`managed-step` and `--resume-turn-key` recovery perform no fresh observation. + +JSON carries capability-owned readback under +`turn_start_capability_hook_dispatch.results[].readback`, with hook id +`issue_fix.pr_wait_observation`: remote read count, duration in milliseconds, +degraded/failure readback, target/cadence diagnostics and observation +`state_written`. Compatibility prequota retains `checks.pr_dependencies`. +Markdown consumes the hook's bounded summary without PR-specific core logic. +Diagnostic lists include at most 16 rows and 16 Todo ids per row, with explicit +omission counts; selection and satisfaction still use complete canonical state. +The 64 KiB hook budget remains unchanged. An uncertain provider/log response +reports `null` reads/effects; rejected post-invocation results also preserve +uncertain effects and invalidate the quota projection cache. Neither case may +certify zero effects. Provider failures do not gate unrelated eligible work. +Observation has no separate model call or business quota debit. It is polling; +network latency and total host costs remain workload-dependent. + +File/SQLite source qualification covers full/thin generated heartbeat commands, +executing guard/native Turn, cross-entrypoint cadence, concurrent invocation, +revocation, actual validated protocol-host work and original-journal replay. +The App/Lark shared context backend reads the same canonical readiness without +polling. This does not qualify an installed App/heartbeat/Lark executor or +automatic recovery of an existing owner's live Goal. Those owners must retain +their actual identity, permission and scheduling readback. Observe, condition +satisfaction, admission and accepted subsequent work are separate evidence. Full decisions retain the selected Task's existing `resume_when`, `resume_ready` and complete `resume_condition`; TurnEnvelope carries the same fields in @@ -69,33 +90,44 @@ signed contents. `managed-step` remains a read-only controller and does not poll `pr_merged:owner/repo#42` 只标识依赖;缺少合并事件不证明 GitHub 仍在等待。 历史交付监督不再豁免这种未观察等待,因果阻塞结算仍要求已注册的 `monitor_changed` 或 `todo_done` 目标及当前回读。 -原生 `heartbeat-prequota --goal-id GOAL --agent-id AGENT` 从完整 Todo 源选择 +注册的 `issue-fix` TurnStart hook 从完整 Todo 源选择 精确的未完成 PR 依赖(含 deferred),每次最多核验四个不同仓库限定的依赖。 -成功和失败均保留 30 分钟核验节奏,重启后仍有效。仓库缺失需要修复,不能按编号猜测。 +成功和失败均按每项依赖保留 30 分钟核验节奏,重启后仍有效;执行入口共用 Goal 内的 +single-flight,并发调用只读回观察进行中,不重复请求。仓库缺失需要修复,不能按编号猜测。 真实合并观察写入既有 `pr_merge`,原条件自然满足。回读与观察共用完整保留的 Goal 日志,在最近事件展示裁剪前由 TS resume owner 保留每项精确依赖的原始合并证明; 超过 500 条无关事件或 256 项其他依赖不能挤走它。list、status、quota 与归档读取 仍保留原展示边界,重启不需要重复远端读取或改写合并时间。离开 open 列表、关闭未合并或 provider 失败不能解除等待。观察不修改 Todo 状态、要求、认领、lease 或 Goal 验收; 已满足条件的 deferred Todo 仍走正常生命周期转换。 -生成的 heartbeat 仅在声明 `--available-capability external_evidence_poll` 时加入 -prequota;只声明 filesystem/shell 不会启用。省略该声明即关闭外部观察。 -显式调用仅授权紧凑元数据读取和本地观察记账,不授权合并、review、远端写入或额度支出。 -通过 `todo list --goal-id GOAL --todo-id TODO --format json` 回读。 - -新的 `turn run-once --execute` 调用在显式声明同一能力后,会在收集本次 governing -decision 前运行既有 PR 依赖观察器。业务额度暂停时仍可满足依赖条件;停止、quota、 -claim、lease 与宿主启动仍由原准入约束。plan、dry run 与 `--resume-turn-key` -恢复不执行新观察;省略能力声明保持原路径。移除声明会停止之后的自动读取,已验证的 -合并事实保留。 - -原生 CLI 的 JSON 返回 `pr_dependency_observation`:远端读取次数、观察耗时毫秒、 -退化/失败读回、既有目标/节奏计划与观察 `state_written`;Markdown 汇总次数、 -节奏与失败。provider -或日志响应不确定时,最后一项为 `null`,失败不能被解释为零本地效果。观察没有单独的 -模型调用或业务额度扣费;沿用每次四项和 30 分钟重试节奏,实际网络延迟及完整宿主成本 -仍依赖工作负载。本切片验证原生 CLI 接入;安装态 heartbeat 声明、App/Lark 调用方采用 -与持续真实调度仍分别验收。 +既有 typed capability owner 从当前 Goal 声明、已记住的宿主观察和本次参数解析 +`external_evidence_poll`;宿主记录的撤销优先于继承授权。owner 已提供权限时,不必给 +每条命令手工补 flag。filesystem/shell、Todo 要求与显式调用 prequota 都不授予网络 +权限。权限来源不可用时只返回失败,不联系 GitHub。移除所有继承授权或记录能力不可用, +即可停止后续自动读取;已验证合并事实保留。 + +执行性的 `quota should-run` 与新的 `turn run-once --execute` 在 governing decision +前组合相同注册 hook;兼容 `heartbeat-prequota` 也调用它。full/thin heartbeat 在宿主 +显式声明轮询时保留原 prequota 命令,执行 guard 仍从当前 owner 解析权限。业务额度暂停 +或业务停止时,仍获授权的观察可以继续;随后工作仍受 stop、quota、claim、lease 与 +启动准入约束。status、list、页面/上下文读取、prompt 生成、plan、dry run、 +`managed-step` 与原 journal 恢复均不执行新观察。 + +JSON 在 `turn_start_capability_hook_dispatch.results[].readback` 返回能力自有读回, +hook id 为 `issue_fix.pr_wait_observation`,包含读取次数、耗时、退化/失败、目标/ +节奏诊断与观察 `state_written`;兼容 prequota 保留 `checks.pr_dependencies`。 +Markdown 消费 hook 的有界摘要,核心 renderer 不持有 PR 业务规则。诊断列表最多 +16 行、每行最多 16 个 Todo id,并声明省略数量;选择和满足判断仍使用完整 canonical +源,64 KiB hook 预算不变。provider/日志响应丢失时,读取/效果为 `null`;执行后 +被契约拒绝的结果也保留效果不确定性并使 quota 投影缓存失效,不能宣称零效果。 +provider 失败不阻挡无关的合法工作。观察不单独调用模型或扣业务额度;它是轮询, +网络延迟与完整宿主成本仍依赖实际负载。 + +File/SQLite 源码验证覆盖 full/thin 生成命令、执行 guard/原生 Turn、跨入口节奏、 +并发、撤销、实际协议宿主的独立校验工作与原 journal 重放。App/Lark 共用上下文后端 +无需轮询即可读取相同 canonical 就绪状态。这不代表已安装 App/heartbeat/Lark +执行器或既有 owner 的真实 Goal 已恢复;这些 owner 仍需以真实身份读回权限和调度。 +观察、条件满足、执行准入与后续验收工作是分别成立的证据。 完整决策保留所选 Task 既有的 `resume_when`、`resume_ready` 与完整 `resume_condition`;TurnEnvelope 将相同字段置于 `action.selected_todo`,纳入 diff --git a/loopx/capabilities/issue_fix/README.md b/loopx/capabilities/issue_fix/README.md index 02a379cded..c00b713d3e 100644 --- a/loopx/capabilities/issue_fix/README.md +++ b/loopx/capabilities/issue_fix/README.md @@ -445,6 +445,23 @@ event does not grant new write authority. [LoopX PR #1883](https://github.com/huangruiteng/loopx/pull/1883) is the implementation and regression evidence for this contract. +`pr_wait_hook.py` registers dependency observation through the existing +TurnStart contract. Executing quota guards, native Turns and compatibility +prequota share the capability's permission resolution, durable per-dependency +cadence and Goal-local single flight. Current Goal declarations and remembered +host capabilities are reused; a Todo requirement never grants networking. +Provider/configuration failure is observable and does not gate independent +business work. The automatic wait observer requires the exact returned +repository URL; redirects do not silently satisfy the original binding. + +The shared core consumes generic hook results; PR-specific target selection, +metadata validation and summaries stay here. Diagnostic rows are bounded after +complete-source selection and declare omitted rows/ids. Status/list/App/Lark +context reads and original-journal recovery consume existing facts without +polling. Installed executor adoption remains separate from source/wheel tests. +See [continuation readback](../../../docs/reference/todo-continuation-readback.md#pr-dependency-observation--pr-依赖观察) +for the entrypoint, permission and failure matrix. + ### 8. Terminal closeout and repeatability At merged/closed state, persist compact lifecycle evidence, close the monitor, diff --git a/loopx/capabilities/issue_fix/pr_wait_hook.py b/loopx/capabilities/issue_fix/pr_wait_hook.py new file mode 100644 index 0000000000..e01cff4197 --- /dev/null +++ b/loopx/capabilities/issue_fix/pr_wait_hook.py @@ -0,0 +1,132 @@ +"""Register PR dependency observation with the shared pre-admission lifecycle.""" +from __future__ import annotations + +import time +from pathlib import Path +from typing import Any + +from ...agent_registry import load_goal_from_registry +from ...control_plane.agents.capability_gate import _evaluate +from ...control_plane.agents.capability_memory import agent_capability_memory +from ...control_plane.capability_hooks import ( + TURN_START_HOOK_RESULT_SCHEMA_VERSION, + TurnStartHookRegistration, +) +from ...control_plane.quota.goal_boundary import declared_available_capabilities +from ...control_plane.todos.contract import normalize_required_capabilities +from .pr_wait_reconcile import reconcile_pr_wait_dependencies + + +HOOK_ID = "issue_fix.pr_wait_observation" + + +def _bounded_readback(readback: dict[str, Any]) -> dict[str, Any]: + """Bound diagnostics only, after the full-source plan and effects finish.""" + def row(value: dict[str, Any]) -> dict[str, Any]: + ids = value.get("todo_ids") + return {**value, "todo_ids": ids[:16], "todo_ids_omitted": max(0, len(ids) - 16)} if isinstance(ids, list) else value + + projected = dict(readback) + projected["results"] = [row(value) for value in readback.get("results", [])] + if isinstance(readback.get("plan"), dict): + plan = dict(readback["plan"]) + coverage = {} + for field in ("targets", "waiting_targets", "unresolved"): + values = plan.get(field, []) + plan[field] = [row(value) for value in values[:16]] + coverage[field] = {"total": len(values), "included": len(plan[field]), + "omitted": max(0, len(values) - 16)} + plan["diagnostic_coverage"] = coverage + projected["plan"] = plan + return projected + + +def _availability(registry_path: Path, runtime_root: Path, goal_id: str, + agent_id: str, available: Any) -> dict[str, Any]: + goal = load_goal_from_registry(registry_path, goal_id) + state = agent_capability_memory(registry_path=registry_path, runtime_root=runtime_root, + goal_id=goal_id, agent_id=agent_id) + return _evaluate("availability", goal=declared_available_capabilities(goal), + runtime=normalize_required_capabilities(available), agent=state) + + +def turn_start_hook(*, registry_path: Path, runtime_root: Path, goal_id: str, + agent_id: str, available: Any, + fetch_timeout_seconds: int = 10) -> TurnStartHookRegistration | None: + # The existing typed owner resolves declarations, remembered host observations + # and revocation. A Todo requirement is never a capability grant. + try: + if "external_evidence_poll" not in _availability( + registry_path, runtime_root, goal_id, agent_id, available, + )["effective"]: + return None + except (ValueError, RuntimeError, OSError, TypeError): + # Register a failure readback, never permission to contact the provider. + pass + + def produce() -> dict[str, Any]: + base = { + "schema_version": TURN_START_HOOK_RESULT_SCHEMA_VERSION, + "hook_id": HOOK_ID, "capability_id": "issue-fix", "phase": "turn_start", + "status": "not_applicable", "observation_count": 0, + "agent_read_required": False, "external_reads_performed": False, + "external_writes_performed": False, "local_private_state_mutated": False, + "private_content_returned": False, "provider_payload_returned": False, + "error_code": None, + } + started = time.monotonic_ns() + try: + # Recheck at invocation, so revocation after registration is effective. + if "external_evidence_poll" not in _availability( + registry_path, runtime_root, goal_id, agent_id, available, + )["effective"]: + return base + except (ValueError, RuntimeError, OSError, TypeError): + return {**base, "status": "unavailable", "error_code": "capability_source_unavailable", + "readback": {"ok": False, "degraded": True, "failure_count": 1, + "external_read_count": 0, "state_written": False, + "summary_lines": ["Observation permission source is unavailable; provider not contacted."]}} + try: + readback = reconcile_pr_wait_dependencies(registry_path=registry_path, + runtime_root_arg=str(runtime_root), goal_id=goal_id, agent_id=agent_id, + fetch_timeout_seconds=fetch_timeout_seconds) + count = readback["external_read_count"] + written = count > 0 + status = "partial" if readback.get("degraded") else "observed" if count else "empty" + error = "dependency_observation_unavailable" if readback.get("degraded") else None + except Exception as exc: # noqa: BLE001 - optional observation is fail-open + # Provider/log response loss cannot certify zero reads or zero writes. + readback = {"ok": False, "degraded": True, "failure_count": 1, + "external_read_count": None, "error_category": type(exc).__name__} + count, written, status, error = None, None, "failed", "dependency_observation_failed" + readback["state_written"] = written + readback["duration_ms"] = (time.monotonic_ns() - started) // 1_000_000 + readback["summary_lines"] = [ + f"PR dependency reads: {count}", + f"PR observation duration_ms: {readback['duration_ms']}", + f"PR dependency observation degraded: {readback.get('degraded')}", + f"PR unavailable dependencies: {readback.get('unavailable_dependency_count')}", + f"PR observation state_written: {written}", + ] + if readback.get("error_category"): + readback["summary_lines"].append(f"PR observation error: {readback['error_category']}") + for item in readback.get("results", []): + readback["summary_lines"].append(f"PR dependency {item['pr_ref']}: {item['state']}") + plan = readback.get("plan") + if isinstance(plan, dict): + readback["summary_lines"].append(f"PR retry interval_seconds: {plan['poll_interval_seconds']}") + observed = {item["pr_ref"] for item in readback.get("results", [])} + for target in plan.get("waiting_targets", []): + if target.get("last_observed_state") == "FAILED" and target["pr_ref"] not in observed: + readback["summary_lines"].append( + f"PR dependency {target['pr_ref']}: FAILED; next_check_at: {target['next_due_at']}") + if len(readback["summary_lines"]) >= 16: + break + return {**base, "status": status, "observation_count": count or 0, + "external_reads_performed": None if count is None else count > 0, + "local_private_state_mutated": written, "error_code": error, + "readback": _bounded_readback(readback)} + + return TurnStartHookRegistration(hook_id=HOOK_ID, capability_id="issue-fix", + requested_read_scope=("goal_agent_registration", "canonical_todos", "github_pr_metadata", "goal_event_log"), + requested_write_scope=("goal_event_log",), producer=produce, max_result_bytes=64 * 1024) diff --git a/loopx/capabilities/issue_fix/pr_wait_reconcile.py b/loopx/capabilities/issue_fix/pr_wait_reconcile.py index 09a0aa9502..c6f73d0c3d 100644 --- a/loopx/capabilities/issue_fix/pr_wait_reconcile.py +++ b/loopx/capabilities/issue_fix/pr_wait_reconcile.py @@ -15,6 +15,7 @@ from ...control_plane.runtime.time import now_utc_iso from ...control_plane.todos.resume_condition import compact_todo_resume_items from ...history import load_registry +from ...file_lock import exclusive_file_lock, LockAcquisitionPolicy, LockAcquireTimeoutError from ...paths import resolve_runtime_root from ...rollout_event_log import ( append_rollout_event, build_rollout_event, rollout_event_log_path, @@ -30,6 +31,27 @@ def reconcile_pr_wait_dependencies( ) -> dict[str, Any]: if agent_id not in registered_agent_ids_from_registry(registry_path, goal_id): raise ValueError("PR dependency observation requires a registered agent") + runtime = (Path(runtime_root_arg).expanduser() if runtime_root_arg + else resolve_runtime_root(load_registry(registry_path), None)) + log = rollout_event_log_path(runtime, goal_id) + # All entrypoints share this Goal-local single flight and the existing + # durable cadence. The lock is coordination, never a second fact ledger. + try: + with exclusive_file_lock(log.with_name(log.name + ".pr-observation"), + policy=LockAcquisitionPolicy.SINGLE_FLIGHT, + agent_id=agent_id, operation="pr_wait_observation"): + return _reconcile_locked(registry_path=registry_path, runtime_root_arg=str(runtime), + goal_id=goal_id, agent_id=agent_id, fetch_timeout_seconds=fetch_timeout_seconds) + except LockAcquireTimeoutError: + return {"ok": True, "degraded": False, "failure_count": 0, "external_read_count": 0, + "observation_in_progress": True, "merged_count": 0, "results": [], + "todo_write_performed": False} + + +def _reconcile_locked( + *, registry_path: Path, runtime_root_arg: str | None, goal_id: str, + agent_id: str, fetch_timeout_seconds: int, +) -> dict[str, Any]: runtime = (Path(runtime_root_arg).expanduser() if runtime_root_arg else resolve_runtime_root(load_registry(registry_path), None)) # Deliberately omit list --agent-id and display limits. The typed owner must diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index b2fae53423..59a827afb2 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -306,14 +306,14 @@ def _dispatch_quota_turn_start_hooks( ) if args.agent_id: from ..capabilities.manager_context import turn_start_hook - from ..control_plane.agents.capability_memory import ( - extend_turn_start_dispatch as extend_capability_memory_dispatch, + from ..control_plane.capabilities.turn_start import ( + extend_turn_start_dispatch as extend_registered_capabilities, ) from ..control_plane.capability_hooks import dispatch_turn_start_hooks from ..history import load_registry from ..paths import resolve_runtime_root root = resolve_runtime_root(load_registry(registry_path), runtime_root_arg, registry_path=registry_path) - dispatch = extend_capability_memory_dispatch( + dispatch = extend_registered_capabilities( dispatch, registry_path=registry_path, runtime_root=root, @@ -341,9 +341,9 @@ def _dispatch_quota_turn_start_hooks( from ..capabilities.explore.turn_context import extend_turn_start_dispatch as extend_explore dispatch = extend_explore(dispatch, registry_path=registry_path, runtime_root=root, goal_id=args.goal_id, agent_id=args.agent_id) - local_private_state_mutated = any( + local_private_state_mutated = bool(dispatch.get("uncertain_local_effects")) or any( isinstance(result, Mapping) - and result.get("local_private_state_mutated") is True + and result.get("local_private_state_mutated") is not False for result in (dispatch.get("results") or []) ) return dispatch, local_private_state_mutated @@ -386,9 +386,6 @@ def _attach_turn_start_hook_dispatch( payload["turn_start_capability_hook_dispatch"] = { key: value for key, value in dispatch.items() if key != "contexts" } - - - def _project_quota_cli_payload( payload: dict[str, object], args: argparse.Namespace, diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index 29dd448f99..9d83b85e63 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -2,7 +2,6 @@ import argparse import shlex -import time from collections.abc import Callable, Mapping from pathlib import Path from typing import Any @@ -41,20 +40,6 @@ FormatSelector = Callable[..., str] -def _attach_pr_dependency_observation( - payload: dict[str, Any], observation: dict[str, Any] | None, -) -> None: - if observation is None: - return - payload["pr_dependency_observation"] = observation - written = observation["state_written"] - effects = payload.setdefault("effects", {}) - if written is True or (written is None and effects.get("state_written") is False): - effects["state_written"] = written - if written is not False and isinstance(payload.get("boundary"), dict): - payload["boundary"]["read_only"] = False - - def handle_turn_command( args: argparse.Namespace, *, @@ -82,7 +67,7 @@ def handle_turn_command( output_format=output_format, print_payload=print_payload, ) payload: dict[str, Any] = {} - pr_dependency_observation: dict[str, Any] | None = None + turn_start_hook_dispatch: dict[str, Any] = {} try: if getattr(args, "todo_id", None) is not None and ( getattr(args, "resume_turn_key", None) @@ -191,7 +176,6 @@ def handle_turn_command( from ..capabilities.agent_turn_recall import run_configured_agent_turn_recall_fail_open from ..capabilities.periodic_report.cadence_runtime import extend_cadence_turn_start_dispatch from .lark_inbox import dispatch_goal_lark_turn_start_hooks - from ..control_plane.todos.contract import normalize_required_capabilities turn_start_hook_dispatch = dispatch_goal_lark_turn_start_hooks( registry_path=registry_path, @@ -202,11 +186,11 @@ def handle_turn_command( turn_start_hook_dispatch = extend_cadence_turn_start_dispatch( turn_start_hook_dispatch, registry_path=registry_path, runtime_root=runtime_root, goal_id=args.goal_id, agent_id=args.agent_id) - from ..control_plane.agents.capability_memory import ( - extend_turn_start_dispatch as extend_capability_memory_dispatch, + from ..control_plane.capabilities.turn_start import ( + extend_turn_start_dispatch as extend_registered_capabilities, ) - turn_start_hook_dispatch = extend_capability_memory_dispatch( + turn_start_hook_dispatch = extend_registered_capabilities( turn_start_hook_dispatch, registry_path=registry_path, runtime_root=runtime_root, @@ -215,33 +199,6 @@ def handle_turn_command( available=args.available_capabilities, ) - # Observe synchronized dependencies before selection and business - # quota, independently of the blocked business Host. - # Plans, dry runs and recovery of an original journal never poll. - if "external_evidence_poll" in normalize_required_capabilities(args.available_capabilities): - from ..capabilities.issue_fix.pr_wait_reconcile import reconcile_pr_wait_dependencies - - observation_started = time.monotonic_ns() - try: - pr_dependency_observation = reconcile_pr_wait_dependencies( - registry_path=registry_path, runtime_root_arg=str(runtime_root), - goal_id=args.goal_id, agent_id=args.agent_id, - ) - pr_dependency_observation["state_written"] = bool( - pr_dependency_observation["external_read_count"] - ) - except Exception as exc: # noqa: BLE001 - optional observation is fail-open - # A lost provider/log response is not proof of zero reads - # or zero local effects. Preserve that uncertainty while - # the ordinary decision owner still checks all admission. - pr_dependency_observation = { - "ok": False, "degraded": True, "failure_count": 1, - "error_category": type(exc).__name__, - "external_read_count": None, "state_written": None, - } - pr_dependency_observation["duration_ms"] = ( - time.monotonic_ns() - observation_started - ) // 1_000_000 from ..capabilities.semantic_preference.agent_preferences import extend_turn_start_dispatch as extend_preferences turn_start_hook_dispatch = extend_preferences(turn_start_hook_dispatch, runtime_root=runtime_root, registry_path=registry_path, goal_id=args.goal_id, agent_id=args.agent_id) @@ -339,17 +296,6 @@ def handle_turn_command( capability_action["command_argv"] = [command[0], "--registry", str(registry_path), "--runtime-root", str(runtime_root), *command[1:]] capability_action["command"] = shlex.join(capability_action["command_argv"]) - if turn_start_hook_dispatch.get("registered_count") or ( - turn_start_hook_dispatch.get("failures") - ): - payload["turn_start_capability_hook_dispatch"] = turn_start_hook_dispatch - if any(isinstance(result, Mapping) and result.get("local_private_state_mutated") is True - for result in turn_start_hook_dispatch.get("results", [])): - # The pure plan builder has no effects, but live preflight - # hooks may journal an inbox or calendar admission. Disclose - # that write without claiming a host turn or report ran. - payload["effects"]["state_written"] = True - payload["boundary"]["read_only"] = False if args.turn_command == "plan": if not args.include_transaction_detail: payload.pop("session", None) @@ -370,7 +316,8 @@ def handle_turn_command( # never run its command as an arbitrary host/shell adapter. payload.update(mode="run_once", status="capability_action_required", execute=bool(args.execute), executed=False) - _attach_pr_dependency_observation(payload, pr_dependency_observation) + from ..control_plane.capabilities.turn_start import attach_turn_start_effects + attach_turn_start_effects(payload, turn_start_hook_dispatch) print_payload(payload, output_format(args), _render_loopx_turn_plan_markdown) return 0 @@ -397,7 +344,8 @@ def handle_turn_command( payload = build_turn_error_payload( payload, exc, turn_command=args.turn_command, ) - _attach_pr_dependency_observation(payload, pr_dependency_observation) + from ..control_plane.capabilities.turn_start import attach_turn_start_effects + attach_turn_start_effects(payload, turn_start_hook_dispatch) renderer = ( _render_loopx_turn_execution_markdown if args.turn_command == "run-once" diff --git a/loopx/cli_commands/turn_rendering.py b/loopx/cli_commands/turn_rendering.py index ac3ff4ef49..9ec7fbc27e 100644 --- a/loopx/cli_commands/turn_rendering.py +++ b/loopx/cli_commands/turn_rendering.py @@ -13,30 +13,15 @@ ) -def _pr_dependency_observation_lines(payload: Mapping[str, object]) -> list[str]: - observation = payload.get("pr_dependency_observation") - if not isinstance(observation, dict): +def _turn_start_observation_lines(payload: Mapping[str, object]) -> list[str]: + dispatch = payload.get("turn_start_capability_hook_dispatch") + if not isinstance(dispatch, Mapping): return [] - lines = [ - f"- PR dependency reads: {observation.get('external_read_count')}", - f"- PR observation duration_ms: {observation.get('duration_ms')}", - f"- PR dependency observation degraded: {observation.get('degraded')}", - f"- PR unavailable dependencies: {observation.get('unavailable_dependency_count')}", - f"- PR observation state_written: {observation.get('state_written')}", - ] - if observation.get("error_category"): - lines.append(f"- PR observation error: {observation['error_category']}") - for result in observation.get("results", []): - lines.append(f"- PR dependency {result['pr_ref']}: {result['state']}" - + (f" ({result['error_category']})" if result.get("error_category") else "")) - plan = observation.get("plan") - if isinstance(plan, dict): - lines.append(f"- PR retry interval_seconds: {plan.get('poll_interval_seconds')}") - observed = {result["pr_ref"] for result in observation.get("results", [])} - retained_failures = [target for target in plan.get("waiting_targets", []) - if target.get("last_observed_state") == "FAILED" and target["pr_ref"] not in observed] - for target in retained_failures[:4]: - lines.append(f"- PR dependency {target['pr_ref']}: FAILED; next_check_at: {target['next_due_at']}") + lines = [] + for result in dispatch.get("results", []): + readback = result.get("readback") + if isinstance(readback, Mapping): + lines.extend(f"- {result['capability_id']}: {line}" for line in readback.get("summary_lines", [])) return lines @@ -99,7 +84,7 @@ def render_loopx_turn_plan_markdown(payload: dict[str, object]) -> str: capability = payload.get("capability_action") if isinstance(payload.get("capability_action"), dict) else {} intent = capability.get("intent") if isinstance(capability.get("intent"), dict) else {} envelope = payload.get("turn_envelope") - observation = payload.get("pr_dependency_observation") + observation = payload.get("turn_start_capability_hook_dispatch") return "\n".join( [ "# LoopX Turn Plan", @@ -107,10 +92,11 @@ def render_loopx_turn_plan_markdown(payload: dict[str, object]) -> str: f"- execution_mode: {host.get('execution_mode')}", f"- route: {route.get('kind')}", f"- would_invoke_host: {route.get('would_invoke_host')}", - ("- side_effects: PR dependency observation" - if isinstance(observation, dict) and observation.get("state_written") is not False + ("- side_effects: registered capability observation" + if isinstance(observation, dict) and (observation.get("uncertain_local_effects") or any( + row.get("local_private_state_mutated") is not False for row in observation.get("results", []))) else "- side_effects: none"), - *_pr_dependency_observation_lines(payload), + *_turn_start_observation_lines(payload), *(["- capability_action: required (not executed)", f"- next_command: {capability.get('command') or intent.get('command')}"] if capability else []), *turn_envelope_budget_warning_lines( @@ -195,7 +181,7 @@ def render_loopx_turn_execution_markdown(payload: dict[str, object]) -> str: f"- host_invoked: {effects.get('host_invoked')}", f"- state_written: {effects.get('state_written')}", f"- quota_spent: {effects.get('quota_spent')}", - *_pr_dependency_observation_lines(payload), + *_turn_start_observation_lines(payload), *([f"- error: {payload['error']}"] if payload.get("error") else []), *([f"- journal_readback: {journal_observation.get('status')}", f"- recorded_effects: {journal_observation.get('recorded_effects')}", diff --git a/loopx/control_plane/capabilities/turn_start.py b/loopx/control_plane/capabilities/turn_start.py new file mode 100644 index 0000000000..03925487b9 --- /dev/null +++ b/loopx/control_plane/capabilities/turn_start.py @@ -0,0 +1,57 @@ +"""Composition of registered built-in observations at executing entrypoints. + +This adapter invokes capability-owned registrations. Selection, permissions, +fact validation and cadence stay with their existing TypeScript owners. +""" +from pathlib import Path +from typing import Any + +from ..capability_hooks import dispatch_turn_start_hooks +from ..todos.contract import normalize_required_capabilities + + +def extend_turn_start_dispatch(dispatch: dict[str, Any], *, registry_path: Path, + runtime_root: Path, goal_id: str, agent_id: str | None, + available: Any, provider_options: dict[str, dict[str, Any]] | None = None) -> dict[str, Any]: + if not agent_id: + return dispatch + from ...capabilities.issue_fix.pr_wait_hook import turn_start_hook as dependencies + from ..agents.capability_memory import turn_start_hook as capability_memory + + factories = [("issue-fix", dependencies)] + if normalize_required_capabilities(available): + factories.append(("agent-capability-memory", capability_memory)) + registrations = [hook for capability, factory in factories if (hook := factory( + registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, + agent_id=agent_id, available=available, + **(provider_options or {}).get(capability, {}), + )) is not None] + if not registrations: + return dispatch + extra = dispatch_turn_start_hooks(registrations) + result = dict(dispatch) + for key in ("results", "required_reads", "failures", "contexts"): + if extra.get(key): + result[key] = list(dispatch.get(key) or []) + list(extra[key]) + for key in ("registered_count", "invoked_count"): + result[key] = int(dispatch.get(key) or 0) + int(extra.get(key) or 0) + if extra.get("uncertain_local_effects"): + result["uncertain_local_effects"] = True + return result + + +def attach_turn_start_effects(payload: dict[str, Any], dispatch: dict[str, Any]) -> None: + """Preserve known and uncertain local effects from every registered hook.""" + if dispatch.get("registered_count") or dispatch.get("failures"): + payload["turn_start_capability_hook_dispatch"] = dispatch + values = [row.get("local_private_state_mutated") for row in dispatch.get("results", [])] + if dispatch.get("uncertain_local_effects"): + values.append(None) + if not any(value is True or value is None for value in values): + return + written = True if any(value is True for value in values) else None + effects = payload.setdefault("effects", {}) + if written is True or effects.get("state_written") is not True: + effects["state_written"] = written + if isinstance(payload.get("boundary"), dict): + payload["boundary"]["read_only"] = False diff --git a/loopx/control_plane/capability_hooks.py b/loopx/control_plane/capability_hooks.py index a9599268df..81747f8730 100644 --- a/loopx/control_plane/capability_hooks.py +++ b/loopx/control_plane/capability_hooks.py @@ -247,6 +247,7 @@ def dispatch_turn_start_hooks( contexts: list[dict[str, Any]] = [] seen_hook_ids: set[str] = set() seen_required_read_commands: set[str] = set() + uncertain_local_effects = False ordered = sorted(registrations or (), key=lambda item: item.hook_id) for registration in ordered: if registration.hook_id in seen_hook_ids: @@ -267,6 +268,7 @@ def dispatch_turn_start_hooks( result = dict(registration.producer()) except Exception: # noqa: BLE001 - capability failures are isolated. failures.append(_hook_failure(registration, error_code="producer_failed")) + uncertain_local_effects |= bool(registration.requested_write_scope) continue try: normalized = effect_runtime_result( @@ -278,11 +280,13 @@ def dispatch_turn_start_hooks( ) except (EffectRuntimeRejected, RuntimeError, TypeError, ValueError): failures.append(_hook_failure(registration, error_code="contract_rejected")) + uncertain_local_effects |= bool(registration.requested_write_scope) continue if not isinstance(normalized, Mapping): failures.append( _hook_failure(registration, error_code="runtime_result_invalid") ) + uncertain_local_effects |= bool(registration.requested_write_scope) continue results.append(dict(normalized)) if normalized.get("agent_read_required") is True and isinstance( @@ -317,6 +321,7 @@ def dispatch_turn_start_hooks( "results": results, "required_reads": required_reads, "failures": failures, + **({"uncertain_local_effects": True} if uncertain_local_effects else {}), **({"contexts": contexts} if contexts else {}), } diff --git a/loopx/control_plane/capability_hooks.ts b/loopx/control_plane/capability_hooks.ts index 41d20d3307..14deda4e7a 100644 --- a/loopx/control_plane/capability_hooks.ts +++ b/loopx/control_plane/capability_hooks.ts @@ -169,6 +169,7 @@ const TURN_START_WRITE_SCOPES = new Set([ "owner_private_inbox", "owner_private_cursor", "provider_message_reaction", + "goal_event_log", ]); const TURN_START_STATUSES = new Set([ "not_applicable", @@ -487,7 +488,18 @@ export function validateTurnStartHookInvocation(input: { }): JsonObject { const registration = validateTurnStartHookRegistration(input.registration); const result = requiredObject(input.result, "turn-start hook result"); - requireExactFields(result, TURN_START_RESULT_FIELDS, "turn-start hook result"); + const fields = new Set(TURN_START_RESULT_FIELDS); + if ("readback" in result) fields.add("readback"); + requireExactFields(result, fields, "turn-start hook result"); + if ("readback" in result) { + const readback = requiredObject(result.readback, "turn-start capability readback"); + if ("summary_lines" in readback) { + const lines = requireStringArray(readback.summary_lines, "turn-start readback summary_lines"); + if (lines.length > 32 || lines.some(line => line.length > 360 || /[\u0000-\u001f\u007f]/.test(line))) { + throw new Error("turn-start readback summary is outside its display budget"); + } + } + } if ( result.schema_version !== TURN_START_HOOK_RESULT_SCHEMA_VERSION || result.hook_id !== registration.hook_id || @@ -521,7 +533,10 @@ export function validateTurnStartHookInvocation(input: { "private_content_returned", "provider_payload_returned", ]) { - if (typeof result[field] !== "boolean") { + const uncertainEffect = result[field] === null && + ["failed", "partial", "unavailable"].includes(status) && + ["external_reads_performed", "local_private_state_mutated"].includes(field); + if (typeof result[field] !== "boolean" && !uncertainEffect) { throw new Error(`turn-start hook ${field} must be boolean`); } } @@ -529,7 +544,7 @@ export function validateTurnStartHookInvocation(input: { throw new Error("turn-start hook cannot return private provider content"); } if ( - result.local_private_state_mutated && + result.local_private_state_mutated !== false && registration.requested_write_scope.length === 0 ) { throw new Error("turn-start hook mutated undeclared local-private state"); @@ -547,7 +562,7 @@ export function validateTurnStartHookInvocation(input: { if (status === "observed" && ( observationCount === 0 || errorCode !== null || - result.agent_read_required !== true + (!result.agent_read_required && !(registration.required_read === null && "readback" in result)) )) { throw new Error("observed turn-start hook result is inconsistent"); } @@ -571,7 +586,7 @@ export function validateTurnStartHookInvocation(input: { if (["partial", "unavailable", "failed"].includes(status) && errorCode === null) { throw new Error("failed turn-start hook result requires error_code"); } - if (status === "partial" && observationCount > 0 && !result.agent_read_required) { + if (status === "partial" && observationCount > 0 && !result.agent_read_required && !(registration.required_read === null && "readback" in result)) { throw new Error("partial turn-start observations require Agent reading"); } if (["unavailable", "failed"].includes(status) && result.agent_read_required) { diff --git a/loopx/heartbeat_prequota.py b/loopx/heartbeat_prequota.py index 70c552a584..289e65fec9 100644 --- a/loopx/heartbeat_prequota.py +++ b/loopx/heartbeat_prequota.py @@ -6,7 +6,9 @@ from .capabilities.issue_fix.pr_gate_reconcile import ( reconcile_acknowledged_issue_fix_pr_reviews, ) -from .capabilities.issue_fix.pr_wait_reconcile import reconcile_pr_wait_dependencies +from .control_plane.capabilities.turn_start import extend_turn_start_dispatch +from .history import load_registry +from .paths import resolve_runtime_root HEARTBEAT_PRE_QUOTA_SCHEMA_VERSION = "heartbeat_pre_quota_v0" @@ -45,13 +47,19 @@ def run_heartbeat_pre_quota( "write_count": 0, } - try: - dependencies = reconcile_pr_wait_dependencies(registry_path=registry_path, - runtime_root_arg=runtime_root_arg, goal_id=goal_id, agent_id=agent_id, - fetch_timeout_seconds=fetch_timeout_seconds) - except Exception as exc: - dependencies = {"ok": False, "degraded": True, "failure_count": 1, - "error_category": type(exc).__name__, "external_read_count": 0} + runtime = resolve_runtime_root(load_registry(registry_path), runtime_root_arg, registry_path=registry_path) + dispatch = extend_turn_start_dispatch({}, registry_path=registry_path, runtime_root=runtime, + goal_id=goal_id, agent_id=agent_id, available=None, + provider_options={"issue-fix": {"fetch_timeout_seconds": fetch_timeout_seconds}}) + # Keep the released command's compatibility projection. Observation itself + # is registered and permission-resolved by the same lifecycle as execution. + failed = next((row for row in dispatch.get("failures", []) + if row.get("capability_id") == "issue-fix"), None) + dependencies = next((row["readback"] for row in dispatch.get("results", []) + if row.get("capability_id") == "issue-fix" and isinstance(row.get("readback"), dict)), + {"ok": not bool(failed), "degraded": bool(failed), "failure_count": 1 if failed else 0, + "external_read_count": None if failed else 0, "state_written": None if failed else False, + "status": "observation_unavailable" if failed else "not_configured_or_unavailable"}) degraded = degraded or bool(dependencies.get("degraded")) failure_count += int(dependencies.get("failure_count") or 0) @@ -66,6 +74,7 @@ def run_heartbeat_pre_quota( "acknowledged_pr_reviews": review_reconciliation, "pr_dependencies": dependencies, }, + "turn_start_capability_hook_dispatch": dispatch, "quota_spend_required": False, "continue_to_quota": True, } diff --git a/tests/capabilities/test_pr_wait_lifecycle.py b/tests/capabilities/test_pr_wait_lifecycle.py new file mode 100644 index 0000000000..902db4f580 --- /dev/null +++ b/tests/capabilities/test_pr_wait_lifecycle.py @@ -0,0 +1,317 @@ +"""Dependency facts cross executing entrypoints without borrowing business authority.""" +from __future__ import annotations + +import json +import sys +import threading +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from loopx.capabilities.issue_fix import pr_lifecycle, pr_wait_reconcile +from loopx.control_plane.agents.capability_memory import agent_capability_memory +from loopx.control_plane.capabilities.turn_start import extend_turn_start_dispatch +from loopx.heartbeat_prequota import run_heartbeat_pre_quota +from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path +from loopx.todos import add_goal_todo, list_goal_todos +from tests.capabilities.test_pr_wait_observation import _invoke_turn, _native_wait_turn, _observation +from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime +from tests.test_loopx_turn_driver import _write_live_fixture + + +def _declare_observation(registry): + config = json.loads(registry.read_text()) + config["goals"][0]["available_capabilities"] = ["external_evidence_poll"] + registry.write_text(json.dumps(config)) + + +def _guard_args(argv, identity): + return [*argv[:argv.index("turn")], "quota", "should-run", "--goal-id", "goal-a", + "--agent-id", "agent-a", "--codex-app", "--turn-instance-id", identity] + + +def _without_invocation_capability(argv): + result = list(argv) + i = result.index("--available-capability") + del result[i:i + 2] + return result + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_shared_cadence_across_quota_native_and_compatibility_prequota(tmp_path, monkeypatch, provider): + registry, runtime, project, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + _declare_observation(registry) + original = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] + clock = ["2026-10-01T01:00:00Z"] + monkeypatch.setattr(pr_wait_reconcile, "now_utc_iso", lambda: clock[0]) + calls = [] + + def fetch(reference, **kwargs): + calls.append(reference) + return {"url": "https://github.com/owner/repo/pull/42", + "state": "OPEN" if len(calls) == 1 else "MERGED", + "mergedAt": None if len(calls) == 1 else "2026-10-01T01:12:00Z"} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + _, guard = _invoke_turn(_guard_args(argv, "configured-guard")) + assert _observation(guard)["external_read_count"] == 1 + assert list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"]["resume_ready"] is False + _, native = _invoke_turn([*_without_invocation_capability(argv), "--turn-instance-id", "inherited-native"]) + assert _observation(native)["external_read_count"] == 0 + compat = run_heartbeat_pre_quota(registry_path=registry, runtime_root_arg=str(runtime), + goal_id="goal-a", agent_id="agent-a") + assert compat["checks"]["pr_dependencies"]["external_read_count"] == 0 + clock[0] = "2026-10-01T01:29:59Z" + _invoke_turn(_guard_args(argv, "not-yet-due")) + assert len(calls) == 1 + clock[0] = "2026-10-01T01:30:00Z" + _, resumed = _invoke_turn(_guard_args(argv, "due-guard")) + assert _observation(resumed)["merged_count"] == 1 + current = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] + assert current["resume_ready"] is True + assert current["resume_condition"]["matched_event_at"] == "2026-10-01T01:12:00Z" + for field in ("text", "status", "claimed_by", "resume_when", "required_write_scopes"): + assert current.get(field) == original.get(field) + events = load_rollout_events(rollout_event_log_path(runtime, "goal-a")) + assert len(calls) == 2 and sum(e["event_kind"] == "pr_merge" for e in events) == 1 + assert not any(e["event_kind"] in {"quota_spend", "todo_complete"} for e in events) + assert not (project / "host-started").exists() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_host_observation_inheritance_revocation_and_existing_fact_retention(tmp_path, monkeypatch, provider): + registry, runtime, _, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + calls = [] + + def fetch(reference, **kwargs): + calls.append(reference) + return {"url": f"https://github.com/{reference['repo']}/pull/{reference['number']}", + "state": "MERGED", "mergedAt": "2026-10-01T00:00:00Z"} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + _, off = _invoke_turn(_guard_args(argv, "unconfigured")) + assert _observation(off) is None and calls == [] + agent_capability_memory(registry_path=registry, runtime_root=runtime, goal_id="goal-a", + agent_id="agent-a", available=["external_evidence_poll"], execute=True) + _, inherited = _invoke_turn(_guard_args(argv, "host-inheritance")) + assert _observation(inherited)["merged_count"] == 1 + next_id = add_goal_todo(registry_path=registry, goal_id="goal-a", role="agent", + text="Observe the next dependency", claimed_by="agent-a", resume_when="pr_merged:owner/repo#43")["todo_id"] + agent_capability_memory(registry_path=registry, runtime_root=runtime, goal_id="goal-a", + agent_id="agent-a", unavailable=["external_evidence_poll"], execute=True) + _declare_observation(registry) # A remembered revocation also defeats inheritance. + _, revoked = _invoke_turn(_guard_args(argv, "revoked")) + assert _observation(revoked) is None and len(calls) == 1 + assert list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=next_id)["todo"]["resume_ready"] is False + assert list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"]["resume_ready"] is True + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_readonly_consumers_do_not_turn_permission_into_observation(tmp_path, monkeypatch, provider): + registry, runtime, _, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + _declare_observation(registry) + calls = [] + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", lambda *args, **kw: calls.append(args)) + prefix = argv[:argv.index("turn")] + _invoke_turn([*prefix, "todo", "list", "--goal-id", "goal-a", "--todo-id", todo_id]) + dry = list(argv) + dry.remove("--execute") + _invoke_turn(dry) + plan = [*prefix, "turn", "plan", "--goal-id", "goal-a", "--agent-id", "agent-a", + "--host", "generic-cli", "--scan-root", str(tmp_path / "project")] + _invoke_turn(plan) + for mode in ("--full", "--thin"): + _invoke_turn([*prefix, "heartbeat-prompt", "--goal-id", "goal-a", "--agent-id", "agent-a", + "--codex-app", mode]) + assert calls == [] + assert list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"]["resume_ready"] is False + assert not any(e["event_kind"] == "pr_merge" for e in load_rollout_events(rollout_event_log_path(runtime, "goal-a"))) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_concurrent_registered_callers_share_one_goal_observation(tmp_path, monkeypatch, provider): + registry, runtime, _, _, _ = _native_wait_turn(tmp_path, monkeypatch, provider) + _declare_observation(registry) + entered, finish = threading.Event(), threading.Event() + calls = [] + + def fetch(reference, **kwargs): + calls.append(reference) + entered.set() + assert finish.wait(10) + return {"url": "https://github.com/owner/repo/pull/42", "state": "OPEN"} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + args = dict(registry_path=registry, runtime_root=runtime, goal_id="goal-a", agent_id="agent-a", available=None) + with ThreadPoolExecutor(max_workers=1) as pool: + first = pool.submit(extend_turn_start_dispatch, {}, **args) + assert entered.wait(10) + try: + second = extend_turn_start_dispatch({}, **args) + observation = next(r["readback"] for r in second["results"] if r["hook_id"] == "issue_fix.pr_wait_observation") + assert observation["observation_in_progress"] is True + assert observation["external_read_count"] == 0 and observation["state_written"] is False + finally: + finish.set() + completed = first.result(timeout=10) + assert len(calls) == 1 + assert next(r["readback"] for r in completed["results"] if r["hook_id"] == "issue_fix.pr_wait_observation")["external_read_count"] == 1 + third = extend_turn_start_dispatch({}, **args) + assert next(r["readback"] for r in third["results"] if r["hook_id"] == "issue_fix.pr_wait_observation")["external_read_count"] == 0 + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_observer_failure_does_not_gate_independent_validated_work(tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture(tmp_path, provider=provider) + _declare_observation(registry) + waiting = add_goal_todo(registry_path=registry, goal_id="loopx-turn-fixture", role="agent", + text="Wait for a separate dependency", claimed_by="codex-fixture", resume_when="pr_merged:owner/repo#42")["todo_id"] + calls = [] + + def unavailable(*args, **kwargs): + calls.append(args) + raise TimeoutError("metadata request unavailable") + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", unavailable) + runner = tmp_path / "protocol_host.py" + runner.write_text( + "import json\nfrom pathlib import Path\n" + "def run_dsh_turn(**kwargs):\n" + " root = Path(kwargs['workspace'])\n" + " (root / 'validated-artifact').write_text('independent work accepted')\n" + " return json.dumps({'result_kind':'validated_progress','classification':'independent_work'," + "'work_evidence':['validated-artifact'],'validation_evidence':['Independent command checks exact artifact']," + "'summary':'Delivered independent fixture work.','delivery_batch_scale':'implementation'," + "'delivery_outcome':'outcome_progress','delivery_boundary':'in_flight_continuation'," + "'recommended_action':'Continue the accepted work.'})\n" + ) + code, result = _invoke_turn(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--host", "dsh", "--dsh-runner", str(runner), "--project", str(project), "--scan-root", str(project), + "--todo-id", "todo_fixture0001", "--turn-instance-id", "independent-work", + "--validation-command-json", json.dumps([sys.executable, "-c", "from pathlib import Path; " + "assert Path('validated-artifact').read_text() == 'independent work accepted'"]), + "--no-global-sync", "--execute"]) + assert code == 0 and result["status"] == "committed", result + assert len(calls) == 1 and _observation(result)["degraded"] is True + assert list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture", todo_id=waiting)["todo"]["resume_ready"] is False + events = load_rollout_events(rollout_event_log_path(runtime, "loopx-turn-fixture")) + assert sum(e["event_kind"] == "quota_spend" for e in events) == 1 + assert not any(e["event_kind"] in {"pr_merge", "todo_complete"} for e in events) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_app_lark_context_backend_reads_shared_facts_without_polling(tmp_path, monkeypatch, provider): + from loopx.capabilities.manager_context.inspection import ManagerInspection + + registry, runtime, _, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + _declare_observation(registry) + calls = [] + + def merged(*args, **kwargs): + calls.append(args) + return {"url": "https://github.com/owner/repo/pull/42", "state": "MERGED", + "mergedAt": "2026-10-01T01:12:00Z"} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", merged) + inspectors = [ManagerInspection(context={"scope": "owner_goal", "goals": [{"goal_id": "goal-a"}]}, + registry_path=registry, runtime_root=runtime, owner_scope=True, scope_valid=lambda: True, + record=lambda result: None, channel_id=channel) for channel in (None, "private-lark-fixture")] + + def read(inspector): + result = inspector.read("loopx_context_read", {"view": "todos", "goal_id": "goal-a", "todo_id": todo_id}) + assert result["ok"] is True and result["included"] == 1, result + return result["rows"][0] + + assert all(read(inspector)["resume_ready"] is False for inspector in inspectors) + assert calls == [] + _invoke_turn(_guard_args(argv, "observe-before-context")) + events = load_rollout_events(rollout_event_log_path(runtime, "goal-a")) + for inspector in inspectors: + row = read(inspector) + assert row["resume_ready"] is True and row["resume_when"] == "pr_merged:owner/repo#42" + assert len(calls) == 1 + assert load_rollout_events(rollout_event_log_path(runtime, "goal-a")) == events + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_lost_observation_response_refreshes_cached_readiness(tmp_path, monkeypatch, provider): + from loopx.capabilities.issue_fix import pr_wait_hook + + registry, runtime, _, todo_id, argv = _native_wait_turn(tmp_path, monkeypatch, provider) + # Capture the old pending projection before permission is configured. + _invoke_turn([*_guard_args(argv, "capture-pending"), "--write-projection-cache"]) + _declare_observation(registry) + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", lambda *a, **kw: { + "url": "https://github.com/owner/repo/pull/42", "state": "MERGED", "mergedAt": "2026-10-01T01:12:00Z"}) + original = pr_wait_hook.reconcile_pr_wait_dependencies + + def response_lost(**kwargs): + original(**kwargs) + raise OSError("response lost after local commit") + + monkeypatch.setattr(pr_wait_hook, "reconcile_pr_wait_dependencies", response_lost) + _, result = _invoke_turn([*_guard_args(argv, "lost-response"), "--use-projection-cache"]) + readback = _observation(result) + assert readback["external_read_count"] is None and readback["state_written"] is None + todo = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] + assert todo["resume_ready"] is True + assert todo["resume_condition"]["matched_event_at"] == "2026-10-01T01:12:00Z" + assert result["turn_start_capability_hook_dispatch"]["results"][0]["external_writes_performed"] is False + + +def test_unavailable_permission_source_never_contacts_provider(tmp_path, monkeypatch): + from loopx.capabilities.issue_fix import pr_wait_hook + + registry, runtime, _, _, _ = _native_wait_turn(tmp_path, monkeypatch, "file") + _declare_observation(registry) + monkeypatch.setattr(pr_wait_hook, "_availability", lambda *a: (_ for _ in ()).throw(OSError("source unavailable"))) + calls = [] + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", lambda *a, **kw: calls.append(a)) + dispatch = extend_turn_start_dispatch({}, registry_path=registry, runtime_root=runtime, + goal_id="goal-a", agent_id="agent-a", available=None) + assert calls == [] and dispatch.get("failures", []) == [] + row = next(r for r in dispatch["results"] if r["hook_id"] == "issue_fix.pr_wait_observation") + assert row["status"] == "unavailable" and row["local_private_state_mutated"] is False + assert row["readback"]["external_read_count"] == 0 + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_full_source_selection_is_independent_of_bounded_hook_diagnostics(tmp_path, monkeypatch, provider): + from loopx.capabilities.issue_fix import pr_wait_hook + + isolate_sqlite_runtime(tmp_path, monkeypatch) + lines = tuple(line for number in range(1, 501) for line in ( + f"- [ ] [P2] Continue dependency {number}.", + f" ")) + _, runtime, registry = _write_live_fixture(tmp_path, provider=provider, extra_agent_todo_lines=lines) + _declare_observation(registry) + calls, full_results = [], [] + original = pr_wait_hook.reconcile_pr_wait_dependencies + + def capture(**kwargs): + result = original(**kwargs) + full_results.append(result) + return result + + def fetch(reference, **kwargs): + calls.append(reference) + return {"url": f"https://github.com/{reference['repo']}/pull/{reference['number']}", "state": "OPEN"} + + monkeypatch.setattr(pr_wait_hook, "reconcile_pr_wait_dependencies", capture) + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + dispatch = extend_turn_start_dispatch({}, registry_path=registry, runtime_root=runtime, + goal_id="loopx-turn-fixture", agent_id="codex-fixture", available=None) + assert dispatch.get("failures", []) == [] + result = next(r for r in dispatch["results"] if r["hook_id"] == "issue_fix.pr_wait_observation") + assert len(calls) == 4 and full_results[0]["plan"]["due_count"] == 500 + assert len(json.dumps(full_results[0]).encode()) > 64 * 1024 + assert len(json.dumps(result).encode()) < 64 * 1024 + assert result["readback"]["plan"]["diagnostic_coverage"]["waiting_targets"] == { + "total": 500, "included": 16, "omitted": 484} + # The next batch comes from complete canonical state, never the diagnostic prefix. + extend_turn_start_dispatch({}, registry_path=registry, runtime_root=runtime, + goal_id="loopx-turn-fixture", agent_id="codex-fixture", available=None) + assert len(calls) == 8 and len({r["number"] for r in calls}) == 8 diff --git a/tests/capabilities/test_pr_wait_observation.py b/tests/capabilities/test_pr_wait_observation.py index 56963c7449..462159f10b 100644 --- a/tests/capabilities/test_pr_wait_observation.py +++ b/tests/capabilities/test_pr_wait_observation.py @@ -16,6 +16,11 @@ ) +def _observation(result): + return next((row.get("readback") for row in result.get("turn_start_capability_hook_dispatch", {}).get("results", []) + if row.get("hook_id") == "issue_fix.pr_wait_observation"), None) + + def _native_wait_turn(tmp_path, monkeypatch, provider, *, claimed_by="agent-a"): isolate_sqlite_runtime(tmp_path, monkeypatch) registry, runtime, _ = promoted_create_fixture( @@ -74,8 +79,8 @@ def fetch(reference, **kwargs): assert result["error"] == "LoopX Turn route is not host executable", result after = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] assert after["resume_ready"] is True, result - assert result["pr_dependency_observation"]["merged_count"] == 1 - assert result["pr_dependency_observation"]["state_written"] is True + assert _observation(result)["merged_count"] == 1 + assert _observation(result)["state_written"] is True assert not (project / "host-started").exists() assert len(calls) == 1 for key in ("text", "status", "claimed_by", "resume_when", "required_write_scopes"): @@ -88,8 +93,8 @@ def fetch(reference, **kwargs): assert code == 1, replay assert replay["error"] == "LoopX Turn route is not host executable", replay assert len(calls) == 1 - assert replay["pr_dependency_observation"]["external_read_count"] == 0 - assert replay["pr_dependency_observation"]["state_written"] is False + assert _observation(replay)["external_read_count"] == 0 + assert _observation(replay)["state_written"] is False assert not (project / "host-started").exists() @@ -149,7 +154,7 @@ def fetch(*args, **kwargs): "--agent-id", "agent-a", "--host", "generic-cli", "--scan-root", str(project), "--available-capability", "external_evidence_poll"] _, result = _invoke_turn(argv) - assert "pr_dependency_observation" not in result + assert _observation(result) is None assert calls == [] assert list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"]["resume_ready"] is False assert not (project / "host-started").exists() @@ -185,7 +190,7 @@ def fetch(*args, **kwargs): failed = outcome not in {"open", "closed"} for number in range(2): _, result = _invoke_turn([*argv, "--turn-instance-id", f"pending-observation-{number}"]) - observation = result["pr_dependency_observation"] + observation = _observation(result) assert observation["degraded"] is failed assert observation["external_read_count"] == (1 if number == 0 else 0) assert observation["unavailable_dependency_count"] == (1 if failed else 0) @@ -217,7 +222,7 @@ def lose_observation_receipt(*args, **kwargs): monkeypatch.setattr(pr_wait_reconcile, "append_rollout_event", lose_observation_receipt) _, result = _invoke_turn(argv) - observation = result["pr_dependency_observation"] + observation = _observation(result) assert observation["ok"] is False and observation["degraded"] is True assert observation["external_read_count"] is None assert observation["state_written"] is None @@ -269,7 +274,7 @@ def test_turn_observer_does_not_take_another_agents_dependency(tmp_path, monkeyp calls = [] monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", lambda *args, **kwargs: calls.append(args)) _, result = _invoke_turn(argv) - assert result["pr_dependency_observation"]["external_read_count"] == 0 + assert _observation(result)["external_read_count"] == 0 after = list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"] assert after["claimed_by"] == "agent-b" and after["resume_ready"] is False assert calls == [] and not (project / "host-started").exists() @@ -292,7 +297,7 @@ def test_observation_does_not_resume_a_stopped_goals_business(tmp_path, monkeypa "url": "https://github.com/owner/repo/pull/42", "state": "MERGED", "mergedAt": "2026-10-01T00:00:00Z", }) _, result = _invoke_turn(argv) - assert result["pr_dependency_observation"]["merged_count"] == 1 + assert _observation(result)["merged_count"] == 1 assert list_goal_todos(registry_path=registry, goal_id="goal-a", todo_id=todo_id)["todo"]["resume_ready"] is True assert json.loads(registry.read_text())["goals"][0]["activation"]["state"] == "stopped" assert not (project / "host-started").exists() @@ -319,7 +324,7 @@ def test_original_journal_recovery_does_not_run_fresh_observation(tmp_path, monk # covered separately by the native Turn driver suite. code, result = _invoke_turn([*argv, "--resume-turn-key", "sha256:" + "a" * 64]) assert code == 0, result - assert calls == [] and "pr_dependency_observation" not in result + assert calls == [] and _observation(result) is None @pytest.mark.parametrize("provider", ["file", "sqlite"]) @@ -349,6 +354,8 @@ def fetch(*args, **kwargs): " counter.write_text(str(int(counter.read_text()) + 1 if counter.exists() else 1))\n" " (root / 'validated-artifact').write_text('continued')\n" " return json.dumps({'result_kind':'validated_progress'," + "'work_evidence':['validated-artifact']," + "'validation_evidence':['Independent command verifies continued artifact']," "'classification':'dependency_followthrough','summary':'Validated the resumed fixture work.'," "'delivery_batch_scale':'implementation','delivery_outcome':'outcome_progress'," "'delivery_boundary':'in_flight_continuation','recommended_action':'Continue the original task.'})\n" @@ -365,7 +372,7 @@ def fetch(*args, **kwargs): code, result = _invoke_turn([*argv, "--turn-instance-id", "adopt-observed-merge"]) assert code == 0 and result["status"] == "committed", result assert (project / "host-count").read_text() == "1" - observation = result["pr_dependency_observation"] + observation = _observation(result) assert observation["merged_count"] == 1 and observation["duration_ms"] >= 0 after = list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture", todo_id="todo_fixture0001")["todo"] assert after["resume_ready"] is True @@ -374,7 +381,7 @@ def fetch(*args, **kwargs): code, replay = _invoke_turn([*argv, "--resume-turn-key", result["resume_turn_key"]]) assert code == 0 and replay["status"] == "committed", replay - assert "pr_dependency_observation" not in replay + assert _observation(replay) is None assert len(calls) == 1 and (project / "host-count").read_text() == "1" events = load_rollout_events(rollout_event_log_path(runtime, "loopx-turn-fixture")) assert sum(event["event_kind"] == "pr_merge" for event in events) == 1 @@ -387,6 +394,9 @@ def test_prequota_observes_merged_dependency_without_open_queue_membership(tmp_p isolate_sqlite_runtime(tmp_path, monkeypatch) registry, runtime, _ = promoted_create_fixture(tmp_path, provider=provider) + config = json.loads(registry.read_text()) + config["goals"][0]["available_capabilities"] = ["external_evidence_poll"] + registry.write_text(json.dumps(config)) created = add_goal_todo(registry_path=registry, goal_id="goal-a", role="agent", text="Continue after the dependency merges", claimed_by="agent-a", resume_when="pr_merged:owner/repo#42") @@ -422,6 +432,9 @@ def test_unsatisfied_and_failed_observations_survive_restart_with_bounded_retry( isolate_sqlite_runtime(tmp_path, monkeypatch) registry, runtime, _ = promoted_create_fixture(tmp_path, provider=provider) + config = json.loads(registry.read_text()) + config["goals"][0]["available_capabilities"] = ["external_evidence_poll"] + registry.write_text(json.dumps(config)) todo_id = add_goal_todo(registry_path=registry, goal_id="goal-a", role="agent", text="Pending dependency", claimed_by="agent-a", resume_when="pr_merged:owner/repo#42")["todo_id"] clock = ["2026-10-01T01:00:00Z"] diff --git a/tests/control_plane/test_pr_wait_consumer_adoption.py b/tests/control_plane/test_pr_wait_consumer_adoption.py index 5ad3d26f71..56a5206ce5 100644 --- a/tests/control_plane/test_pr_wait_consumer_adoption.py +++ b/tests/control_plane/test_pr_wait_consumer_adoption.py @@ -15,7 +15,7 @@ from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path from loopx.todos import add_goal_todo, list_goal_todos -from tests.capabilities.test_pr_wait_observation import _invoke_turn +from tests.capabilities.test_pr_wait_observation import _invoke_turn, _observation from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime from tests.test_loopx_turn_driver import _write_live_fixture @@ -29,6 +29,9 @@ def test_selected_merge_proof_survives_capture_and_actual_host_adoption( project, runtime, registry = _write_live_fixture( tmp_path, provider=provider, todo_metadata_extra="resume_when=pr_merged:other/repo#42", ) + config = json.loads(registry.read_text()) + config["goals"][0]["available_capabilities"] = ["external_evidence_poll"] + registry.write_text(json.dumps(config)) requirement = "Preserve the accepted requirement. " * 22 + "Require the terminal independent artifact marker." todo_id = add_goal_todo( registry_path=registry, goal_id="loopx-turn-fixture", role="agent", claimed_by="codex-fixture", @@ -95,6 +98,8 @@ def capture(decision, **kwargs): " counter.write_text(str(int(counter.read_text()) + 1 if counter.exists() else 1))\n" " (root / 'validated-artifact').write_text('terminal independent artifact marker')\n" " return json.dumps({'result_kind':'validated_progress','classification':'dependency_followthrough'," + "'work_evidence':['validated-artifact']," + "'validation_evidence':['Independent command verifies the terminal artifact marker']," "'summary':'Validated the original dependency-bound fixture work.','delivery_batch_scale':'implementation'," "'delivery_outcome':'outcome_progress','delivery_boundary':'in_flight_continuation'," "'recommended_action':'Continue the original accepted work.'})\n" @@ -107,7 +112,7 @@ def capture(decision, **kwargs): "--no-global-sync", "--execute"] code, result = _invoke_turn(argv) assert code == 0 and result["status"] == "committed", json.dumps(result, indent=2) - assert result["pr_dependency_observation"]["external_read_count"] == (2 if observer_entry == "native" else 0) + assert _observation(result)["external_read_count"] == (2 if observer_entry == "native" else 0) assert len(captured) == 1 full = captured[0] envelope = build_turn_envelope(full) diff --git a/tests/control_plane/test_pr_wait_history.py b/tests/control_plane/test_pr_wait_history.py index 48ecf7f77c..a8ac608e78 100644 --- a/tests/control_plane/test_pr_wait_history.py +++ b/tests/control_plane/test_pr_wait_history.py @@ -18,7 +18,8 @@ def test_merged_wait_stays_ready_after_history_growth_and_restart(tmp_path, monk isolate_sqlite_runtime(tmp_path, monkeypatch) registry, runtime, _ = promoted_create_fixture(tmp_path, provider=provider) data = json.loads(registry.read_text()) - data["goals"][0].update(domain="code", adapter={"kind": "manual"}) + data["goals"][0].update(domain="code", adapter={"kind": "manual"}, + available_capabilities=["external_evidence_poll"]) registry.write_text(json.dumps(data)) todo_id = add_goal_todo(registry_path=registry, goal_id="goal-a", role="agent", text="Continue the original requirement", claimed_by="agent-a", diff --git a/tests/control_plane/test_turn_start_capability_hooks.py b/tests/control_plane/test_turn_start_capability_hooks.py index 3bb797d3a6..1bb33e40fe 100644 --- a/tests/control_plane/test_turn_start_capability_hooks.py +++ b/tests/control_plane/test_turn_start_capability_hooks.py @@ -70,6 +70,17 @@ def test_provider_shape_mismatch_is_not_misclassified_as_empty() -> None: dispatch = dispatch_turn_start_hooks([_hook(lambda: malformed)]) assert dispatch["results"] == [] + # Validation happens after the provider was invoked. An invalid response + # cannot certify that its declared local effects did not occur. + assert dispatch["uncertain_local_effects"] is True + from loopx.control_plane.capabilities.turn_start import attach_turn_start_effects + from loopx.cli_commands.turn_rendering import render_loopx_turn_plan_markdown + + payload = {"ok": True, "effects": {"state_written": False}, "boundary": {"read_only": True}} + attach_turn_start_effects(payload, dispatch) + assert payload["effects"]["state_written"] is None + assert payload["boundary"]["read_only"] is False + assert "side_effects: registered capability observation" in render_loopx_turn_plan_markdown(payload) assert dispatch["failures"] == [ { "hook_id": "operator_inbox.turn_start_sync", diff --git a/tests/control_plane_ts/capability_hooks.test.ts b/tests/control_plane_ts/capability_hooks.test.ts index 9cbd03160b..6f1d712161 100644 --- a/tests/control_plane_ts/capability_hooks.test.ts +++ b/tests/control_plane_ts/capability_hooks.test.ts @@ -464,6 +464,31 @@ function turnStartResult(overrides: Record = {}) { }; } +test("registered fact observations carry bounded readback without an unread message", () => { + const registration = turnStartRegistration({required_read: null, requested_write_scope: ["goal_event_log"]}); + const result = turnStartResult({agent_read_required: false, + readback: {external_read_count: 2, state_written: true, summary_lines: ["Observed two dependencies."]}}); + assert.deepEqual(validateTurnStartHookInvocation({registration, result}), result); + assert.throws(() => validateTurnStartHookInvocation({registration, + result: {...result, readback: {summary_lines: ["not\\nseparate instructions".replace("\\n", "\n")]}}}), /display budget/); + assert.throws(() => validateTurnStartHookInvocation({registration, + result: {...result, readback: {body: "x".repeat(17 * 1024)}}}), /budget/); + assert.throws(() => validateTurnStartHookInvocation({registration, + result: {...result, private_content_returned: true}}), /private provider content/); +}); + +test("failed observations retain uncertain effects only inside declared scope", () => { + const registration = turnStartRegistration({required_read: null, requested_write_scope: ["goal_event_log"]}); + const result = turnStartResult({status: "failed", observation_count: 0, agent_read_required: false, + external_reads_performed: null, local_private_state_mutated: null, + error_code: "observation_response_lost", readback: {external_read_count: null, state_written: null}}); + assert.deepEqual(validateTurnStartHookInvocation({registration, result}), result); + assert.throws(() => validateTurnStartHookInvocation({registration: {...registration, requested_write_scope: []}, result}), /undeclared local-private state/); + assert.throws(() => validateTurnStartHookInvocation({registration, result: {...result, external_writes_performed: null}}), /must be boolean/); + assert.throws(() => validateTurnStartHookInvocation({registration, + result: {...result, status: "observed", observation_count: 1, error_code: null}}), /must be boolean/); +}); + test("turn-start read commands retain explicit routes with a bounded byte budget", () => { const read = {kind: "capability_intent", reason: "Read due report facts", ordering: "before_work"}; const command = `loopx --registry /${"project/".repeat(35)}registry.json --runtime-root /${"runtime/".repeat(35)}state periodic-report consume-pending --goal-id example --agent-id reporter`; From 9f6f58b7f33e9c59b74e311fd7b87439b33e8174 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 18:16:09 +0800 Subject: [PATCH 4/4] refactor(capabilities): compose observations outside the control plane Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/capabilities/issue_fix/pr_wait_hook.py | 22 +++++++++++-------- .../capabilities/turn_start.py | 8 +++---- loopx/cli_commands/quota.py | 2 +- loopx/cli_commands/turn.py | 6 ++--- loopx/heartbeat_prequota.py | 2 +- tests/capabilities/test_pr_wait_lifecycle.py | 2 +- .../test_turn_start_capability_hooks.py | 2 +- 7 files changed, 24 insertions(+), 20 deletions(-) rename loopx/{control_plane => }/capabilities/turn_start.py (88%) diff --git a/loopx/capabilities/issue_fix/pr_wait_hook.py b/loopx/capabilities/issue_fix/pr_wait_hook.py index e01cff4197..74689d73dc 100644 --- a/loopx/capabilities/issue_fix/pr_wait_hook.py +++ b/loopx/capabilities/issue_fix/pr_wait_hook.py @@ -6,18 +6,17 @@ from typing import Any from ...agent_registry import load_goal_from_registry -from ...control_plane.agents.capability_gate import _evaluate -from ...control_plane.agents.capability_memory import agent_capability_memory +from ...control_plane.agents.capability_memory import ( + agent_capability_memory, resolve_agent_capabilities, +) from ...control_plane.capability_hooks import ( TURN_START_HOOK_RESULT_SCHEMA_VERSION, TurnStartHookRegistration, ) -from ...control_plane.quota.goal_boundary import declared_available_capabilities -from ...control_plane.todos.contract import normalize_required_capabilities from .pr_wait_reconcile import reconcile_pr_wait_dependencies -HOOK_ID = "issue_fix.pr_wait_observation" +PR_WAIT_OBSERVATION_HOOK_ID = "issue_fix.pr_wait_observation" def _bounded_readback(readback: dict[str, Any]) -> dict[str, Any]: @@ -46,8 +45,13 @@ def _availability(registry_path: Path, runtime_root: Path, goal_id: str, goal = load_goal_from_registry(registry_path, goal_id) state = agent_capability_memory(registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, agent_id=agent_id) - return _evaluate("availability", goal=declared_available_capabilities(goal), - runtime=normalize_required_capabilities(available), agent=state) + # Resolve from the same scoped owner as quota, with the successfully read + # state supplied explicitly; an unavailable permission source cannot fall + # back to an inherited grant for provider IO. + return resolve_agent_capabilities(status_payload={"registry": str(registry_path), + "runtime_root": str(runtime_root)}, goal_id=goal_id, + agent_identity={"agent_id": agent_id}, item=goal, project_asset={}, + available=available, state=state) def turn_start_hook(*, registry_path: Path, runtime_root: Path, goal_id: str, @@ -67,7 +71,7 @@ def turn_start_hook(*, registry_path: Path, runtime_root: Path, goal_id: str, def produce() -> dict[str, Any]: base = { "schema_version": TURN_START_HOOK_RESULT_SCHEMA_VERSION, - "hook_id": HOOK_ID, "capability_id": "issue-fix", "phase": "turn_start", + "hook_id": PR_WAIT_OBSERVATION_HOOK_ID, "capability_id": "issue-fix", "phase": "turn_start", "status": "not_applicable", "observation_count": 0, "agent_read_required": False, "external_reads_performed": False, "external_writes_performed": False, "local_private_state_mutated": False, @@ -127,6 +131,6 @@ def produce() -> dict[str, Any]: "local_private_state_mutated": written, "error_code": error, "readback": _bounded_readback(readback)} - return TurnStartHookRegistration(hook_id=HOOK_ID, capability_id="issue-fix", + return TurnStartHookRegistration(hook_id=PR_WAIT_OBSERVATION_HOOK_ID, capability_id="issue-fix", requested_read_scope=("goal_agent_registration", "canonical_todos", "github_pr_metadata", "goal_event_log"), requested_write_scope=("goal_event_log",), producer=produce, max_result_bytes=64 * 1024) diff --git a/loopx/control_plane/capabilities/turn_start.py b/loopx/capabilities/turn_start.py similarity index 88% rename from loopx/control_plane/capabilities/turn_start.py rename to loopx/capabilities/turn_start.py index 03925487b9..e6c8e2f083 100644 --- a/loopx/control_plane/capabilities/turn_start.py +++ b/loopx/capabilities/turn_start.py @@ -6,8 +6,8 @@ from pathlib import Path from typing import Any -from ..capability_hooks import dispatch_turn_start_hooks -from ..todos.contract import normalize_required_capabilities +from ..control_plane.capability_hooks import dispatch_turn_start_hooks +from ..control_plane.todos.contract import normalize_required_capabilities def extend_turn_start_dispatch(dispatch: dict[str, Any], *, registry_path: Path, @@ -15,8 +15,8 @@ def extend_turn_start_dispatch(dispatch: dict[str, Any], *, registry_path: Path, available: Any, provider_options: dict[str, dict[str, Any]] | None = None) -> dict[str, Any]: if not agent_id: return dispatch - from ...capabilities.issue_fix.pr_wait_hook import turn_start_hook as dependencies - from ..agents.capability_memory import turn_start_hook as capability_memory + from .issue_fix.pr_wait_hook import turn_start_hook as dependencies + from ..control_plane.agents.capability_memory import turn_start_hook as capability_memory factories = [("issue-fix", dependencies)] if normalize_required_capabilities(available): diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index 59a827afb2..a73c03c9b9 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -306,7 +306,7 @@ def _dispatch_quota_turn_start_hooks( ) if args.agent_id: from ..capabilities.manager_context import turn_start_hook - from ..control_plane.capabilities.turn_start import ( + from ..capabilities.turn_start import ( extend_turn_start_dispatch as extend_registered_capabilities, ) from ..control_plane.capability_hooks import dispatch_turn_start_hooks diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index 9d83b85e63..1bf4b3019e 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -186,7 +186,7 @@ def handle_turn_command( turn_start_hook_dispatch = extend_cadence_turn_start_dispatch( turn_start_hook_dispatch, registry_path=registry_path, runtime_root=runtime_root, goal_id=args.goal_id, agent_id=args.agent_id) - from ..control_plane.capabilities.turn_start import ( + from ..capabilities.turn_start import ( extend_turn_start_dispatch as extend_registered_capabilities, ) @@ -316,7 +316,7 @@ def handle_turn_command( # never run its command as an arbitrary host/shell adapter. payload.update(mode="run_once", status="capability_action_required", execute=bool(args.execute), executed=False) - from ..control_plane.capabilities.turn_start import attach_turn_start_effects + from ..capabilities.turn_start import attach_turn_start_effects attach_turn_start_effects(payload, turn_start_hook_dispatch) print_payload(payload, output_format(args), _render_loopx_turn_plan_markdown) return 0 @@ -344,7 +344,7 @@ def handle_turn_command( payload = build_turn_error_payload( payload, exc, turn_command=args.turn_command, ) - from ..control_plane.capabilities.turn_start import attach_turn_start_effects + from ..capabilities.turn_start import attach_turn_start_effects attach_turn_start_effects(payload, turn_start_hook_dispatch) renderer = ( _render_loopx_turn_execution_markdown diff --git a/loopx/heartbeat_prequota.py b/loopx/heartbeat_prequota.py index 289e65fec9..81cb1f1ef2 100644 --- a/loopx/heartbeat_prequota.py +++ b/loopx/heartbeat_prequota.py @@ -6,7 +6,7 @@ from .capabilities.issue_fix.pr_gate_reconcile import ( reconcile_acknowledged_issue_fix_pr_reviews, ) -from .control_plane.capabilities.turn_start import extend_turn_start_dispatch +from .capabilities.turn_start import extend_turn_start_dispatch from .history import load_registry from .paths import resolve_runtime_root diff --git a/tests/capabilities/test_pr_wait_lifecycle.py b/tests/capabilities/test_pr_wait_lifecycle.py index 902db4f580..adc6489611 100644 --- a/tests/capabilities/test_pr_wait_lifecycle.py +++ b/tests/capabilities/test_pr_wait_lifecycle.py @@ -10,7 +10,7 @@ from loopx.capabilities.issue_fix import pr_lifecycle, pr_wait_reconcile from loopx.control_plane.agents.capability_memory import agent_capability_memory -from loopx.control_plane.capabilities.turn_start import extend_turn_start_dispatch +from loopx.capabilities.turn_start import extend_turn_start_dispatch from loopx.heartbeat_prequota import run_heartbeat_pre_quota from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path from loopx.todos import add_goal_todo, list_goal_todos diff --git a/tests/control_plane/test_turn_start_capability_hooks.py b/tests/control_plane/test_turn_start_capability_hooks.py index 1bb33e40fe..626dbe0c65 100644 --- a/tests/control_plane/test_turn_start_capability_hooks.py +++ b/tests/control_plane/test_turn_start_capability_hooks.py @@ -73,7 +73,7 @@ def test_provider_shape_mismatch_is_not_misclassified_as_empty() -> None: # Validation happens after the provider was invoked. An invalid response # cannot certify that its declared local effects did not occur. assert dispatch["uncertain_local_effects"] is True - from loopx.control_plane.capabilities.turn_start import attach_turn_start_effects + from loopx.capabilities.turn_start import attach_turn_start_effects from loopx.cli_commands.turn_rendering import render_loopx_turn_plan_markdown payload = {"ok": True, "effects": {"state_written": False}, "boundary": {"read_only": True}}