diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md index 362729b842..5e6dab5b5a 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md @@ -626,6 +626,18 @@ fresh Turn settling once. Boundary-length valid results retain their behavior. This is result-transport correctness; it does not qualify model adoption, undo Host work already performed, or close the three convergence streams. +Turn completion now carries an exact text commitment from the admitted full +Todo source to the existing atomic terminal owner. Changed requirements refuse +completion; note-only and unrelated Todo changes remain valid. Provider CAS +fences concurrent replacement, and historical receipt recovery precedes fresh +source checks. Saved Turns without full-source proof refuse new completion; +already accepted legacy receipts retain their fingerprints and recover first. +File/SQLite CLI and shared File/SQLite/PostgreSQL conformance +cover refusal and recovery. Final no-followup retains its existing settlement +order: a terminal refusal preserves any earlier writeback/spend receipts. +This qualifies task-body completion safety, not settlement-wide source atomicity, +installed host adoption, live models or the remaining convergence acceptance. + The first ordinary-CLI consolidation replaces an exactly identical `heartbeat_recommendation.blocked_priority_fallback` mirror with `blocked_priority_fallback_ref="$.blocked_priority_fallback"`. The complete root 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 09cf650df5..0d891d47bb 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 @@ -431,6 +431,14 @@ File/SQLite 原生 CLI 修正后的新 Turn 只结算一次。边界长度的合法结果保持原行为。这只证明结果传输 正确性，不证明模型采用、撤销 Host 已执行工作或关闭原三批收敛验收。 +Turn 完成写入现将已准入 Todo 全文的精确摘要交给既有原子终态 owner。正文要求 +变化拒绝完成，单独备注或其他 Todo 变化仍有效；provider CAS 阻止并发替换，历史 +receipt 恢复先于当前来源检查。已保存 Turn 缺少全文来源证明时拒绝新完成写入， +已接受的旧 receipt 则保留原指纹并优先恢复。真实 File/SQLite CLI 和共享 File/SQLite/PostgreSQL +conformance 覆盖拒绝与恢复。最终 no-followup 保留既有结算顺序：终态拒绝保留此前 +已经提交的 writeback/spend receipt。本批只证明任务正文的完成安全性，不证明整个 +结算的来源原子性、安装态宿主采用、真实模型或其余收敛验收完成。 + 首个普通 CLI 整合将完全相同的 `heartbeat_recommendation.blocked_priority_fallback` 副本替换为 `blocked_priority_fallback_ref="$.blocked_priority_fallback"`。根字段的完整事实、 diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index 2a958d2fe4..69c09872c3 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/loopx/cli_commands/turn_run_once.py @@ -54,7 +54,7 @@ from .turn_dsh_host import build_dsh_host_runner, build_dsh_operation_host_runner from .turn_rendering import build_turn_error_payload from .turn_post_writeback import drain_committed_turn_post_writeback_hooks -from .turn_todo_writeback import write_turn_validated_completion +from .turn_todo_writeback import captured_todo_text_sha256, write_turn_validated_completion EXACT_SETTLEMENT_READBACK_NOT_FOUND = ( "exact settlement readback unexpectedly returned not-found" @@ -122,6 +122,9 @@ def execute_turn_run_once( else {} ) selected_todo = selected_turn_todo(envelope) + expected_todo_text_sha256 = captured_todo_text_sha256( + envelope, todo_id=str(selected_todo.get("todo_id") or "") + ) transaction = ( payload.get("transaction") if isinstance(payload.get("transaction"), Mapping) @@ -350,6 +353,7 @@ def todo_completion( goal_id=args.goal_id, todo_id=todo_id, completion_turn_key=settlement_identity.turn_instance_id, + expected_todo_text_sha256=expected_todo_text_sha256, evidence=( "LoopX Turn validated completion: " + str(result.get("summary") or result["classification"]) @@ -359,6 +363,8 @@ def todo_completion( completion_delivery_workspace=completion_delivery_workspace, completion_validation_workspace_path=delivery_workspace_path, ) + if completion.get("ok") is not True: + return completion # Project the continuation the Todo lifecycle durably recorded, # never a host-normalized continuation. Contradictory or # dangling durable state fails closed before any further diff --git a/loopx/cli_commands/turn_todo_writeback.py b/loopx/cli_commands/turn_todo_writeback.py index 7c31930132..651b6f3578 100644 --- a/loopx/cli_commands/turn_todo_writeback.py +++ b/loopx/cli_commands/turn_todo_writeback.py @@ -1,10 +1,64 @@ """Todo writeback adapters that preserve the Turn's effective runtime root.""" + from __future__ import annotations from pathlib import Path from typing import Any, Mapping +from hashlib import sha256 from ..control_plane.todos.mutation_api import complete_goal_todo +from ..control_plane.coordination.local_authority import ( + LocalCoordinationAuthorityRejection, +) + + +def captured_todo_text_sha256( + envelope: Mapping[str, Any], *, todo_id: str +) -> str | None: + """Resolve admitted full-source references; hash bytes without rereading live state. + + Older journal envelopes have no verification marker and retain their + original completion request. New producers mark source validation before + body deduplication, so a display preview alone can never become the source. + """ + action = envelope.get("action") or {} + selected = action.get("selected_todo") or {} + if "text_source_verified" not in selected: + return None + if ( + selected["text_source_verified"] is not True + or selected.get("todo_id") != todo_id + ): + raise ValueError("Turn task source does not match its selected Todo") + context = envelope.get("work_context") or {} + if not isinstance(context, Mapping) or not isinstance( + context.get("sources", []), list + ): + raise ValueError("Turn completion requires its admitted full task source") + records = [] + for source in context.get("sources", []): + if not isinstance(source, Mapping) or source.get("source") != "selected_todo": + continue + content = source.get("content") + record = content.get("todo") if isinstance(content, Mapping) else None + if ( + isinstance(record, Mapping) + and record.get("todo_id") == todo_id + and isinstance(record.get("text"), str) + ): + records.append(record["text"]) + if len(records) > 1: + raise ValueError("Turn task source is ambiguous") + text = records[0] if records else None + if text is None and ( + not context or context.get("selected_todo_ref") == "selected_todo" + ): + text = selected.get("text") + if text is None and selected.get("text_ref") == "action.recommended_action": + text = action.get("recommended_action") + if not isinstance(text, str) or not text: + raise ValueError("Turn completion requires its admitted full task source") + return sha256(text.encode("utf-8")).hexdigest() def write_turn_validated_completion( @@ -17,24 +71,32 @@ def write_turn_validated_completion( evidence: str, note: str, agent_id: str | None, + expected_todo_text_sha256: str | None = None, completion_delivery_workspace: Mapping[str, Any] | None = None, completion_validation_workspace_path: Path | None = None, ) -> dict[str, Any]: """Complete one validated Todo under the effective runtime root.""" - return complete_goal_todo( - registry_path=registry_path, - goal_id=goal_id, - todo_id=todo_id, - role="agent", - completion_turn_key=completion_turn_key, - completion_identity_source="turn_settlement", - completion_delivery_workspace=completion_delivery_workspace, - completion_validation_workspace_path=completion_validation_workspace_path, - evidence=evidence, - note=note, - agent_id=agent_id, - project=None, - dry_run=False, - runtime_root_arg=runtime_root_arg, - ) + try: + return complete_goal_todo( + registry_path=registry_path, + goal_id=goal_id, + todo_id=todo_id, + role="agent", + completion_turn_key=completion_turn_key, + completion_identity_source="turn_settlement", + expected_todo_text_sha256=expected_todo_text_sha256, + require_captured_todo_text=True, + completion_delivery_workspace=completion_delivery_workspace, + completion_validation_workspace_path=completion_validation_workspace_path, + evidence=evidence, + note=note, + agent_id=agent_id, + project=None, + dry_run=False, + runtime_root_arg=runtime_root_arg, + ) + except LocalCoordinationAuthorityRejection as exc: + # A typed owner rejection proves that this attempt did not write. Lost + # responses and unavailable authority still raise for receipt recovery. + return {**exc.payload, "ok": False, "appended": False, "completed": False} diff --git a/loopx/control_plane/coordination/local_authority_runtime.ts b/loopx/control_plane/coordination/local_authority_runtime.ts index 52b5b9de31..72f4b1d7d0 100644 --- a/loopx/control_plane/coordination/local_authority_runtime.ts +++ b/loopx/control_plane/coordination/local_authority_runtime.ts @@ -1322,6 +1322,10 @@ export async function terminalLifecycleLocalCoordinationTodo( const researchEvidence = new ResearchTerminalEvidenceHost(root, goalId, requireJsonObject(input.registry_source, "registry source")); try { return {...await executeCoordinationTodoTerminalLifecycle(store, { + ...(input.expected_todo_text_sha256 === undefined ? {} : { + expected_todo_text_sha256: input.expected_todo_text_sha256 === null ? null + : requireAuthorityStoreId(input.expected_todo_text_sha256, "expected Todo text digest"), + }), validation_source_provider_revision: input.validation_source_provider_revision == null ? null : requireAuthorityStoreId(input.validation_source_provider_revision, "validation source provider revision"), validation_declaration_sha256: input.validation_declaration_sha256 == null diff --git a/loopx/control_plane/coordination/todo_terminal_lifecycle.ts b/loopx/control_plane/coordination/todo_terminal_lifecycle.ts index 9a5656266e..06c1771705 100644 --- a/loopx/control_plane/coordination/todo_terminal_lifecycle.ts +++ b/loopx/control_plane/coordination/todo_terminal_lifecycle.ts @@ -96,6 +96,9 @@ type TodoRole = typeof TODO_ROLES[number]; type CompletionIdentitySource = typeof COMPLETION_IDENTITY_SOURCES[number]; interface CoordinationTodoTerminalLifecycleBaseInput { + /** Captured task digest. Null refuses fresh completion without source proof; + * omission preserves unscoped callers. Historical receipts recover first. */ + readonly expected_todo_text_sha256?: string | null; readonly review_basis?: {readonly provider_revision: string; readonly registry_sha256: string}; /** Presence selects source-bound validation; null means no effect issued yet. */ readonly validation_source_provider_revision?: string | null; @@ -368,6 +371,10 @@ function validateSuccessorSemantics( function normalizeTerminalInput( raw: CoordinationTodoTerminalLifecycleInput, ): CoordinationTodoTerminalLifecycleInput { + if (raw.expected_todo_text_sha256 != null && + (typeof raw.expected_todo_text_sha256 !== "string" || !BARE_SHA256_PATTERN.test(raw.expected_todo_text_sha256))) { + throw new AuthorityStoreProtocolError("expected Todo text commitment must be a SHA-256 digest"); + } if (raw.review_basis !== undefined) { const basis = canonicalAuthorityObject(raw.review_basis, "terminal review basis"); if (Object.keys(basis).some(key => !["provider_revision", "registry_sha256"].includes(key)) || @@ -500,6 +507,11 @@ function terminalRequestSha(input: CoordinationTodoTerminalLifecycleInput): stri // command binds both its approved snapshot and its complete prose intent. ...(input.review_basis === undefined ? {} : {review_basis: input.review_basis, note: input.note, evidence: input.evidence, reason: input.reason}), + // Null means the native Turn lacks source proof. Preserve the historical + // fingerprint so an accepted old receipt can still recover before refusal. + ...(input.expected_todo_text_sha256 == null ? {} : { + expected_todo_text_sha256: input.expected_todo_text_sha256, + }), ...(input.user_update === undefined ? {} : {user_update: { patch: input.user_update.patch, clear_fields: input.user_update.clear_fields, planning_intent: input.user_update.planning_intent ?? {}, @@ -1296,6 +1308,20 @@ export async function executeCoordinationTodoTerminalLifecycle( "decision_rejection", ); } + // Exact operation receipts above recover historical effects before this + // fresh-source check. The provider CAS below also prevents a source change + // between this comparison and the terminal write. + if (input.expected_todo_text_sha256 === null) { + return terminalFailure("todo_text_source_unavailable", + "Turn completion lacks its admitted full task source; reread the task and run a fresh Turn", + {todo_id: input.todo_id}, "decision_rejection"); + } + if (input.expected_todo_text_sha256 !== undefined && + (typeof todo.text !== "string" || createHash("sha256").update(todo.text, "utf8").digest("hex") !== input.expected_todo_text_sha256)) { + return terminalFailure("todo_text_changed", + "Todo requirements changed after Turn admission; reread the task and run a fresh Turn", + {todo_id: input.todo_id}, "decision_rejection"); + } if (todo.archive_state !== "active") { return terminalFailure( "todo_archived", diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 8a12038a46..59dff9d2cb 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -237,6 +237,8 @@ function selectedTodo(payload: JsonObject, recommendedAction: string | null): Js ]) { if (source[field] !== null && source[field] !== undefined) compact[field] = source[field]; } + const context = object(object(object(payload.interaction_contract).agent_channel).work_context); + if (context.selected_todo_text_verified === true) compact.text_source_verified = true; // Work declarations can end in acceptance/stop conditions. Keep their exact // text; a display prefix or case-folded recommendation cannot stand in for it. const rendered = scalarString(source.text, "selected_todo.text"); @@ -743,7 +745,10 @@ function actionProjection(payload: JsonObject, protocolActionFields: JsonObject) ); const context = object(interaction.agent_context); if (Object.keys(context).length > 0) projection.agent_context = structuredClone(context); - const workContext = object(object(interaction.agent_channel).work_context); + // Source verification belongs to the selected task above. Preserve one carrier while + // retaining the existing context elision and historical packet signatures. + const {selected_todo_text_verified: _taskTextVerified, ...workContext} = + object(object(interaction.agent_channel).work_context); const selectedTodoContextIsAlreadyInAction = workContext.complete === true && workContext.selected_todo_ref === "selected_todo" && Object.keys(workContext).every(field => diff --git a/loopx/control_plane/todos/mutation_api.py b/loopx/control_plane/todos/mutation_api.py index 6e71589bab..5a56c05ff7 100644 --- a/loopx/control_plane/todos/mutation_api.py +++ b/loopx/control_plane/todos/mutation_api.py @@ -449,6 +449,8 @@ def complete_goal_todo( completion_turn_key: str | None = None, completion_identity_source: str | None = None, terminal_review_basis: Mapping[str, Any] | None = None, + expected_todo_text_sha256: str | None = None, + require_captured_todo_text: bool = False, completion_delivery_workspace: Mapping[str, Any] | None = None, completion_validation_workspace_path: Path | None = None, task_lease_idempotency_key: str | None = None, diff --git a/loopx/control_plane/todos/provider_terminal_lifecycle.py b/loopx/control_plane/todos/provider_terminal_lifecycle.py index 7532f09238..de3259de58 100644 --- a/loopx/control_plane/todos/provider_terminal_lifecycle.py +++ b/loopx/control_plane/todos/provider_terminal_lifecycle.py @@ -125,6 +125,8 @@ def route_terminal_call(command: str, call: Mapping[str, Any]) -> dict[str, Any] call.get("completion_identity_source") if complete else None ), review_basis=call.get("terminal_review_basis"), + expected_todo_text_sha256=call.get("expected_todo_text_sha256"), + require_captured_todo_text=call.get("require_captured_todo_text") is True, completion_delivery_workspace=( call.get("completion_delivery_workspace") if complete else None ), @@ -275,6 +277,8 @@ def terminal_canonical_todo_if_promoted( self_merged: bool, dry_run: bool, review_basis: Mapping[str, Any] | None = None, + expected_todo_text_sha256: str | None = None, + require_captured_todo_text: bool = False, project: Path | None = None, state_file: Path | None = None, ) -> dict[str, Any] | None: @@ -356,6 +360,8 @@ def terminal_canonical_todo_if_promoted( request = { "schema_version": _TERMINAL_REQUEST_SCHEMA, **({"review_basis": dict(review_basis)} if review_basis is not None else {}), + **({"expected_todo_text_sha256": expected_todo_text_sha256} + if expected_todo_text_sha256 is not None or require_captured_todo_text else {}), "validation_source_provider_revision": None, "runtime_root": str(runtime_root.expanduser().resolve(strict=False)), "goal_id": goal_id, diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts index ce0331038c..cf2d2b0f2e 100644 --- a/loopx/control_plane/work_items/interaction_contract.ts +++ b/loopx/control_plane/work_items/interaction_contract.ts @@ -426,6 +426,7 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject { const pending: JsonObject[] = [], sources: JsonObject[] = [], failures: JsonObject[] = []; let selectedTodoRef = false; let selectedTodoAuthority: string | null = null; + let selectedTodoTextVerified = false; for (const value of request.required_reads) { const read = requireJsonObject(value, "required read"); const matches = results.filter(result => result.command === read.command); @@ -472,6 +473,12 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject { // full read instead of guessing which prose is current Goal authority. pending.push(read); } else { + if (read.source === "selected_todo" && typeof selectedTodoRecord?.text === "string") { + // Bind completion to the admitted full source, before body deduplication + // or display compaction. Provider-wide revisions also change for notes + // and unrelated work; those do not replace this task's requirements. + selectedTodoTextVerified = true; + } if (read.source === "selected_todo" && typeof selected.text === "string" && selectedTodoRecord?.text === selected.text && selectedTodoRecord?.todo_id === selected.todo_id) { @@ -518,6 +525,7 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject { const includeUserTodos = users && !users.error_code && (!Array.isArray(users.todos) || users.todos.length > 0 || users.authority_read != null); return {required_reads: pending, work_context: {complete: failures.length === 0, + ...(selectedTodoTextVerified ? {selected_todo_text_verified: true} : {}), ...(selectedTodoRef ? {selected_todo_ref: "selected_todo"} : {}), ...(selectedTodoAuthority ? {selected_todo_authority: selectedTodoAuthority} : {}), ...(sources.length ? {sources} : {}), diff --git a/loopx/presentation/renderers/turn_envelope_markdown.py b/loopx/presentation/renderers/turn_envelope_markdown.py index e1b437cbcd..9a3e77ed34 100644 --- a/loopx/presentation/renderers/turn_envelope_markdown.py +++ b/loopx/presentation/renderers/turn_envelope_markdown.py @@ -6,7 +6,8 @@ def work_context_lines(context: dict[str, Any]) -> list[str]: """Complete source content once; arbitrary source text stays JSON-escaped.""" - if set(context) <= {"complete", "selected_todo_ref", "selected_todo_authority", "instruction"}: + # The machine-only source witness does not add another instruction body. + if set(context) <= {"complete", "selected_todo_ref", "selected_todo_authority", "instruction", "selected_todo_text_verified"}: lines = [ "- work_context: " f"complete={context.get('complete')} " diff --git a/tests/control_plane/test_turn_completion_source.py b/tests/control_plane/test_turn_completion_source.py new file mode 100644 index 0000000000..64d373671f --- /dev/null +++ b/tests/control_plane/test_turn_completion_source.py @@ -0,0 +1,478 @@ +"""A validated result may complete only the task body that its Turn admitted.""" + +from __future__ import annotations + +import contextlib +import io +import json +import subprocess +from hashlib import sha256 +from pathlib import Path + +import pytest + +from loopx.cli import main +from loopx.control_plane.turn_driver import executor +from loopx.cli_commands.turn_todo_writeback import captured_todo_text_sha256 +from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime +from tests.control_plane import canonical_authority_fixture +from tests.test_loopx_turn_driver import ( + _completion_host_and_validation_scripts, + _turn_journal, + _turn_run_once_completion_argv, + _write_live_fixture, +) + + +@pytest.mark.parametrize( + "carrier", + ["canonical_detail", "deduplicated_text", "elided_context", "recommendation_ref"], +) +def test_captured_todo_hash_uses_complete_admitted_bytes(carrier: str) -> None: + text = "保留空白与全部要求。 " * 50 + " Keep the final checksum.\n" + selected = {"todo_id": "todo_example", "text_source_verified": True, "text": text} + envelope = { + "action": {"selected_todo": selected}, + "work_context": {"complete": True, "selected_todo_ref": "selected_todo"}, + } + if carrier == "canonical_detail": + selected["text"] = "A display prefix" + envelope["work_context"] = { + "complete": True, + "sources": [ + { + "source": "selected_todo", + "content": {"todo": {"todo_id": "todo_example", "text": text}}, + } + ], + } + elif carrier == "elided_context": + envelope.pop("work_context") + elif carrier == "recommendation_ref": + selected.pop("text") + selected["text_ref"] = "action.recommended_action" + envelope["action"]["recommended_action"] = text + assert ( + captured_todo_text_sha256(envelope, todo_id="todo_example") + == sha256(text.encode("utf-8")).hexdigest() + ) + + +@pytest.mark.parametrize( + "failure", + [ + "missing_full_source", + "ambiguous_source", + "wrong_identity", + "false_witness", + "unknown_reference", + ], +) +def test_captured_todo_hash_refuses_unproven_previews(failure: str) -> None: + selected = { + "todo_id": "todo_example", + "text_source_verified": True, + "text": "A display prefix", + } + envelope = { + "action": {"selected_todo": selected}, + "work_context": {"complete": True}, + } + if failure == "ambiguous_source": + record = { + "source": "selected_todo", + "content": { + "todo": {"todo_id": "todo_example", "text": "Full requirement"} + }, + } + envelope["work_context"]["sources"] = [record, record] + elif failure == "wrong_identity": + selected["todo_id"] = "todo_other" + elif failure == "false_witness": + selected["text_source_verified"] = False + elif failure == "unknown_reference": + envelope.pop("work_context") + selected.pop("text") + selected["text_ref"] = "unregistered.reference" + with pytest.raises(ValueError): + captured_todo_text_sha256(envelope, todo_id="todo_example") + + +def test_context_witness_preserves_legacy_requests_and_markdown_obligations() -> None: + from loopx.presentation.renderers.turn_envelope_markdown import work_context_lines + + envelope = { + "action": {"selected_todo": {"todo_id": "todo_example", "text": "Old preview"}} + } + assert captured_todo_text_sha256(envelope, todo_id="todo_example") is None + context = { + "complete": True, + "selected_todo_ref": "selected_todo", + "instruction": "Read every requirement; stop before release.", + } + assert work_context_lines(context) == work_context_lines( + {**context, "selected_todo_text_verified": True} + ) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("accepted_before_upgrade", [False, True]) +def test_native_legacy_turn_recovers_history_but_cannot_complete_without_source( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + provider: str, + accepted_before_upgrade: bool, +) -> None: + import loopx.cli_commands.turn_run_once as command + from loopx.control_plane.work_items import context_readback + from loopx.control_plane.todos.mutation_api import complete_goal_todo + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture(tmp_path, provider=provider) + prefix = [ + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "--format", + "json", + ] + actual_context = context_readback.attach_work_context + + def legacy_context(*args, **kwargs): + actual_context(*args, **kwargs) + # Reproduce the supported pre-witness producer, including its original + # action signature rather than editing a signed journal after capture. + args[0]["interaction_contract"]["agent_channel"]["work_context"].pop( + "selected_todo_text_verified", None + ) + + monkeypatch.setattr(context_readback, "attach_work_context", legacy_context) + host, validator = _completion_host_and_validation_scripts() + argv = _turn_run_once_completion_argv(project, runtime, registry, host, validator) + actual_write = command.write_turn_validated_completion + results = [] + + def recover_old_writer(**kwargs): + assert kwargs["expected_todo_text_sha256"] is None + if not results: + # Old writers omitted the commitment. Commit through the actual + # lifecycle API, then lose its response before the journal checkpoint. + legacy = { + k: v for k, v in kwargs.items() if k != "expected_todo_text_sha256" + } + result = complete_goal_todo( + role="agent", completion_identity_source="turn_settlement", **legacy + ) + assert result["ok"] and result["provider_status"] == "applied", result + results.append(result) + raise OSError("Synthetic old-writer response loss") + result = actual_write(**kwargs) + results.append(result) + return result + + if accepted_before_upgrade: + monkeypatch.setattr( + command, "write_turn_validated_completion", recover_old_writer + ) + code, result = _cli([], *argv) + assert code == 1, result + if not accepted_before_upgrade: + assert result["result_kind"] == "writeback_failed", result + assert "lacks its admitted full task source" in result["reason"] + assert result["effects"] == { + "host_invoked": True, + "state_written": False, + "quota_spent": False, + "scheduler_acknowledged": False, + } + assert _todo(prefix)["todo"]["status"] == "open" + return + assert result["error"] == "Synthetic old-writer response loss", result + journal = _turn_journal(runtime) + assert ( + "text_source_verified" + not in journal["plan"]["turn_envelope"]["action"]["selected_todo"] + ) + resumed_code, resumed = _cli( + [], *argv[:-1], "--resume-turn-key", str(journal["turn_key"]), "--execute" + ) + assert resumed_code == 0 and resumed["status"] == "committed", resumed + assert resumed["effects"]["host_invoked"] is False + assert results[1]["provider_status"] == "replayed", results[1] + assert _todo(prefix)["todo"]["status"] == "done" + + +def _cli(prefix: list[str], *args: str) -> tuple[int, dict]: + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = main([*prefix, *args]) + return code, json.loads(output.getvalue()) + + +def _todo(prefix: list[str], todo_id: str = "todo_fixture0001") -> dict: + code, result = _cli( + prefix, + "todo", + "list", + "--goal-id", + "loopx-turn-fixture", + "--todo-id", + todo_id, + ) + assert code == 0, result + return result + + +def _update( + prefix: list[str], operation: str, *fields: str, todo_id: str = "todo_fixture0001" +) -> None: + current = _todo(prefix, todo_id) + code, result = _cli( + prefix, + "todo", + "update", + "--goal-id", + "loopx-turn-fixture", + "--todo-id", + todo_id, + "--agent-id", + "codex-fixture", + "--update-operation-id", + operation, + "--update-expected-provider-revision", + current["authority_read"]["provider_revision"], + *fields, + ) + assert code == 0 and result["changed"], json.dumps(result) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("change", ["requirements", "note", "other_todo"]) +@pytest.mark.parametrize("no_followup", [False, True]) +def test_native_turn_completion_checks_full_source_and_preserves_unrelated_changes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + provider: str, + change: str, + no_followup: bool, +) -> None: + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, + provider=provider, + todo_metadata_extra="no_followup=true" if no_followup else "", + extra_agent_todo_lines=( + "- [ ] [P3] Inspect an independent artifact.", + " ", + ) + if change == "other_todo" + else (), + ) + prefix = [ + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "--format", + "json", + ] + # Changing only a distant suffix must not disappear into a 500-character + # selected-task preview, a display title, or normalized whitespace. + original = ( + "[P0] Export public rows. " + "Retain each row. " * 50 + "Output version one." + ) + replacement = ( + original.removesuffix("Output version one.") + "Output a verified checksum." + ) + _update(prefix, "declare-full-requirements", "--text", original) + host, validator = _completion_host_and_validation_scripts() + host = host.replace( + "request = json.load(sys.stdin)", + """request = json.load(sys.stdin) +pathlib.Path("request.json").write_text(json.dumps(request), encoding="utf-8") +calls = pathlib.Path("host-calls.txt") +calls.write_text(calls.read_text() + "x" if calls.exists() else "x")""", + ) + argv = _turn_run_once_completion_argv(project, runtime, registry, host, validator) + real_validator = executor._run_task_validator + + def change_after_real_validation(*args, **kwargs): + receipt = real_validator(*args, **kwargs) + assert receipt["ok"], receipt + if change == "requirements": + _update(prefix, "replace-requirements", "--text", replacement) + elif change == "note": + _update(prefix, "record-observation", "--note", "A concurrent observation.") + else: + _update( + prefix, + "independent-work", + "--text", + "[P3] Inspect another independent artifact.", + todo_id="todo_fixture0002", + ) + return receipt + + monkeypatch.setattr(executor, "_run_task_validator", change_after_real_validation) + code, result = _cli([], *argv) + request = json.loads((project / "request.json").read_text()) + captured_todo = request["turn_envelope"]["action"]["selected_todo"] + assert captured_todo["todo_id"] == "todo_fixture0001" + assert captured_todo["text_source_verified"] is True + current = _todo(prefix)["todo"] + if change != "requirements": + assert code == 0 and result["status"] == "committed", result + assert current["status"] == "done" + assert result["effects"]["quota_spent"] is True + return + + assert code == 1, result + failure_kind = "terminal_closeout_failed" if no_followup else "writeback_failed" + assert result["status"] == "failed" and result["result_kind"] == failure_kind, ( + json.dumps(result) + ) + assert result["effects"] == { + "host_invoked": True, + # Final no-followup commits writeback and spend before its terminal + # phase. A terminal refusal preserves that accepted settlement prefix. + "state_written": no_followup, + "quota_spent": no_followup, + "scheduler_acknowledged": False, + } + assert current["status"] == "open" and current["text"] == replacement + assert (project / "completion-artifact.txt").read_text() == "completed" + journal = _turn_journal(runtime) + assert "durable_writeback" not in journal.get("effect_attempts", {}) + # A cached old result remains refused without another Host call or spend. + monkeypatch.setattr(executor, "_run_task_validator", real_validator) + retry = [ + *argv[:-1], + "--resume-turn-key", + str(journal["turn_key"]), + "--retry-failed-turn", + "--execute", + ] + retry_code, retried = _cli([], *retry) + assert retry_code == 1 and retried["result_kind"] == failure_kind, json.dumps( + retried + ) + assert retried["effects"]["host_invoked"] is False + assert retried["effects"]["quota_spent"] is False + assert (project / "host-calls.txt").read_text() == "x" + assert _todo(prefix)["todo"]["status"] == "open" + + # The newly admitted task needs a matching artifact and independent check. + fresh_host = host.replace( + 'write_text("completed",', 'write_text("checksum:verified",' + ) + fresh_validator = validator.replace('== "completed"', '== "checksum:verified"') + fresh = _turn_run_once_completion_argv( + project, runtime, registry, fresh_host, fresh_validator + ) + fresh_code, accepted = _cli( + [], *fresh[:-1], "--turn-instance-id", "fresh-checksum-turn", "--execute" + ) + assert fresh_code == 0 and accepted["status"] == "committed", accepted + assert accepted["effects"]["quota_spent"] is True + assert _todo(prefix)["todo"]["status"] == "done" + assert (project / "completion-artifact.txt").read_text() == "checksum:verified" + replay_code, replay = _cli( + [], + *fresh[:-1], + "--resume-turn-key", + str(accepted["receipt"]["turn_key"]), + "--execute", + ) + assert replay_code == 0 and replay["replayed"] is True, replay + assert not any(replay["effects"].values()) + assert (project / "host-calls.txt").read_text() == "xx" + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_native_turn_recovers_accepted_completion_before_checking_new_text( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + provider: str, +) -> None: + import loopx.cli_commands.turn_run_once as command + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture(tmp_path, provider=provider) + prefix = [ + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "--format", + "json", + ] + host, validator = _completion_host_and_validation_scripts() + argv = _turn_run_once_completion_argv(project, runtime, registry, host, validator) + actual_completion = command.write_turn_validated_completion + completions: list[dict] = [] + + def lose_response_after_commit(**kwargs): + completion = actual_completion(**kwargs) + completions.append(completion) + if len(completions) == 1: + assert completion["ok"] and completion["status"] == "done", completion + # The public authoring path intentionally forbids replacing a done + # task. Inject a later synthetic provider head in this disposable + # store to challenge the recovery adapter, retaining its receipt. + module = ( + Path(canonical_authority_fixture.__file__).resolve().parents[2] + / "loopx/control_plane/coordination" + ) + script = ( + f"import {{openLocalAuthorityStore}} from {json.dumps((module / 'local_authority_provider.ts').as_uri())};" + f"import {{prepareCoordinationProjectionCommit}} from {json.dumps((module / 'coordination_projection.ts').as_uri())};" + "let text='';for await(const chunk of process.stdin)text+=chunk;const input=JSON.parse(text);" + "const store=await openLocalAuthorityStore(input.root,input.goal);const head=await store.loadAuthority();" + "const todo=head.head.todos.find(row=>row.todo_id==='todo_fixture0001');" + "const result=await store.commitAuthority(prepareCoordinationProjectionCommit({" + "goal_id:input.goal,operation_id:'fixture-later-source',expected_provider_revision:head.provider_revision," + "projection:head.head,mutations:[{kind:'todo_upsert',todo:{...todo,text:input.text}}]}));" + "if(result.status!=='applied')throw new Error(JSON.stringify(result));" + ) + changed = subprocess.run( + [ + "node", + "--no-warnings", + "--experimental-strip-types", + "--input-type=module", + "-e", + script, + ], + input=json.dumps( + { + "root": str(runtime), + "goal": "loopx-turn-fixture", + "text": "[P0] A later requirement after acceptance.", + } + ), + text=True, + capture_output=True, + timeout=30, + ) + assert changed.returncode == 0, changed.stderr + raise OSError("Synthetic response loss after canonical acceptance") + return completion + + monkeypatch.setattr( + command, "write_turn_validated_completion", lose_response_after_commit + ) + code, failed = _cli([], *argv) + assert ( + code == 1 + and failed.get("error") == "Synthetic response loss after canonical acceptance" + ), failed + journal = _turn_journal(runtime) + resumed_code, resumed = _cli( + [], *argv[:-1], "--resume-turn-key", str(journal["turn_key"]), "--execute" + ) + assert resumed_code == 0 and resumed["status"] == "committed", resumed + assert resumed["effects"]["host_invoked"] is False + assert completions[1]["provider_status"] == "replayed", completions[1] + assert _todo(prefix)["todo"]["text"] == "[P0] A later requirement after acceptance." diff --git a/tests/control_plane_ts/interaction_contract.test.ts b/tests/control_plane_ts/interaction_contract.test.ts index 1d062d0cce..2de9ec1050 100644 --- a/tests/control_plane_ts/interaction_contract.test.ts +++ b/tests/control_plane_ts/interaction_contract.test.ts @@ -284,6 +284,7 @@ test("selected Todo references its admitted body only after an exact detail read const context = projected.work_context as JsonObject; assert.equal(context.complete, true); assert.equal(context.selected_todo_ref, "selected_todo"); + assert.equal(context.selected_todo_text_verified, true); const sources = context.sources as JsonObject[]; assert.equal(sources.length, 1); assert.equal(sources[0].source, "selected_todo"); @@ -313,6 +314,7 @@ test("bounded selected text validates its full source snapshot and preserves exa const context = projected.work_context as JsonObject; assert.equal(context.complete, true); assert.equal(context.selected_todo_ref, undefined); + assert.equal(context.selected_todo_text_verified, true); const source = (context.sources as JsonObject[])[0]; const sourceContent = source.content as JsonObject; assert.equal((sourceContent.todo as JsonObject).text, body); @@ -322,6 +324,7 @@ test("bounded selected text validates its full source snapshot and preserves exa source_results: [{command: read.command, content: {matched: true, todo: {...task, text: body + " Updated after selection."}}}]}); assert.equal((changed.work_context as JsonObject).complete, false); + assert.equal((changed.work_context as JsonObject).selected_todo_text_verified, undefined); assert.deepEqual(changed.required_reads, [read]); }); diff --git a/tests/control_plane_ts/terminal_source_conformance.ts b/tests/control_plane_ts/terminal_source_conformance.ts index 62fdd6eb2e..f24e85dd01 100644 --- a/tests/control_plane_ts/terminal_source_conformance.ts +++ b/tests/control_plane_ts/terminal_source_conformance.ts @@ -1,6 +1,7 @@ /** One mixed production-scale graph; independent expectations for review, * validation, lease retirement and receipt recovery on every real provider. */ import assert from "node:assert/strict"; +import {createHash} from "node:crypto"; import {executeCoordinationTodoUpdate} from "../../loopx/control_plane/coordination/todo_update.ts"; import test from "node:test"; import type {AuthorityStore} from "../../loopx/control_plane/coordination/authority_store.ts"; @@ -27,6 +28,111 @@ const validation = {schema_version: "issue_fix_validation_command_v0", command_l export function registerTerminalSourceConformance(provider: string, factory: AuthorityStoreConformanceFactory): void { for (const schema of ["legacy", "native"] as const) { + test(`${provider}: Turn completion binds exact task text through CAS and historical recovery (${schema})`, async t => { + const {store, contender} = await factory(t); + const goal = "turn-text-source"; + const body = "Retain all rows. ".repeat(50) + "最后保留 checksum v1."; + const digest = (text: string) => createHash("sha256").update(text, "utf8").digest("hex"); + const target: JsonObject = {todo_id: "todo_turn_source", role: "agent", status: "open", done: false, + archive_state: "active", text: body, task_class: "advancement_task", claimed_by: "agent-a"}; + const other = {...target, todo_id: "todo_other", text: "Inspect an independent artifact."}; + await store.commitAuthority({operation_id: "seed", expected_provider_revision: null, + next_projection: authorityProjectionFixture(goal, [target, other], [], schema, {handoff_mode: "soft_claim"}), + events: [], receipts: []}); + const request: CoordinationTodoTerminalLifecycleInput = { + goal_id: goal, todo_id: String(target.todo_id), expected_role: "agent", command: "complete", + actor_agent_id: "agent-a", registered_agents: ["agent-a"], lifecycle_grants: [], + authority_reason: null, decision_outcome: null, + operation_identity: {kind: "explicit", operation_id: "complete-turn-source"}, + expected_todo_text_sha256: digest(body), + lease_idempotency_key: null, lease_expected_version: null, allow_user_gate_auto_acquire: false, + requested_no_followup: false, requested_completion_turn_key: "turn-source", + requested_completion_identity_source: "turn_settlement", linked_successor_todo_ids: [], + successor_intents: [], note: "Verified artifact", evidence: "Independent artifact check passed", + reason: null, clear_claim: false, validation_declaration: null, validation_receipt: null, + completion_policy_request: null, dry_run: false, now: new Date("2026-09-05T23:00:00Z"), + }; + let head = await loaded(store); + const missing = await execute(store, {...request, expected_todo_text_sha256: null}); + assert.equal(missing.reason_code, "todo_text_source_unavailable"); + assert.deepEqual(await loaded(store), head); + assert.equal((await store.readReceipt(request.operation_identity.operation_id)).status, "missing"); + // Old accepted requests had no commitment. Explicit missing proof must + // retain their fingerprint for receipt recovery, but cannot write anew. + const legacy = {...request, todo_id: String(other.todo_id), expected_todo_text_sha256: undefined, + operation_identity: {kind: "explicit" as const, operation_id: "legacy-accepted"}}; + assert.equal((await execute(store, legacy)).status, "applied"); + head = await loaded(store); + assert.equal((await execute(contender, {...legacy, expected_todo_text_sha256: null}, + async () => {throw new Error("Legacy receipt must recover before missing-source rejection");})).status, "replayed"); + assert.deepEqual(await loaded(store), head); + const storedTarget = (head.head.todos as JsonObject[]).find(todo => todo.todo_id === target.todo_id)!; + const storedOther = (head.head.todos as JsonObject[]).find(todo => todo.todo_id === other.todo_id)!; + const invalid = await execute(store, {...request, expected_todo_text_sha256: "invalid"}); + assert.equal(invalid.reason_code, "invalid_coordination_todo_terminal_lifecycle"); + assert.deepEqual(await loaded(store), head); + // Notes and other work advance the global revision without replacing the + // requirement this result proves. Preserve both concurrent changes. + await contender.commitAuthority(prepareCoordinationProjectionCommit({goal_id: goal, + operation_id: "unrelated-writes", expected_provider_revision: head.provider_revision, + projection: head.head, mutations: [ + {kind: "todo_upsert", todo: {...storedTarget, note: "Concurrent observation"}}, + {kind: "todo_upsert", todo: {...storedOther, text: "Different independent requirement"}}, + ]})); + const committed = await execute(store, request); + assert.equal(committed.status, "applied", JSON.stringify(committed)); + assert.equal((await execute(store, {...request, expected_todo_text_sha256: null})).reason_code, + "coordination_operation_identity_mismatch"); + head = await loaded(store); + const accepted = (head.head.todos as JsonObject[]).find(todo => todo.todo_id === target.todo_id)!; + assert.equal(accepted.status, "done"); + const nextText = body.replace("checksum v1", "checksum v2"); + await contender.commitAuthority(prepareCoordinationProjectionCommit({goal_id: goal, + operation_id: "new-requirement", expected_provider_revision: head.provider_revision, + projection: head.head, mutations: [{kind: "todo_upsert", todo: { + ...accepted, status: "open", done: false, text: nextText, + }}]})); + head = await loaded(store); + const originalReceipt = await store.readReceipt(request.operation_identity.operation_id); + const replay = await execute(contender, request, + async () => {throw new Error("Receipt recovery must precede current-source checks");}); + assert.equal(replay.status, "replayed", JSON.stringify(replay)); + assert.deepEqual(await loaded(store), head); + assert.deepEqual(await store.readReceipt(request.operation_identity.operation_id), originalReceipt); + assert.equal((await execute(store, {...request, expected_todo_text_sha256: digest(nextText)})).reason_code, + "coordination_operation_identity_mismatch"); + const stale = {...request, operation_identity: {kind: "explicit" as const, operation_id: "stale-completion"}}; + assert.equal((await execute(store, stale)).reason_code, "todo_text_changed"); + assert.deepEqual(await loaded(store), head); + assert.equal((await store.readReceipt("stale-completion")).status, "missing"); + // The source changes after the body check but before the provider write. + // That write must lose CAS rather than completing the replacement task. + let raced = false; + const racingStore = new Proxy(store, {get(owner, key) { + if (key === "commitAuthority") return async (...args: Parameters) => { + assert.equal(raced, false); + raced = true; + const current = await loaded(contender); + const todo = (current.head.todos as JsonObject[]).find(row => row.todo_id === target.todo_id)!; + const changed = await contender.commitAuthority(prepareCoordinationProjectionCommit({goal_id: goal, + operation_id: "during-completion-cas", expected_provider_revision: current.provider_revision, + projection: current.head, mutations: [{kind: "todo_upsert", todo: {...todo, text: nextText + " Keep all footnotes."}}]})); + assert.equal(changed.status, "applied"); + return owner.commitAuthority(...args); + }; + const member = Reflect.get(owner, key); + return typeof member === "function" ? member.bind(owner) : member; + }}); + const racingRequest = {...stale, expected_todo_text_sha256: digest(nextText), + operation_identity: {kind: "explicit" as const, operation_id: "racing-completion"}}; + const conflict = await execute(racingStore, racingRequest); + assert.equal(raced, true); + assert.equal(conflict.status, "conflict", JSON.stringify(conflict)); + assert.equal((await store.readReceipt("racing-completion")).status, "missing"); + assert.equal((await execute(store, racingRequest)).reason_code, "todo_text_changed"); + const afterRace = await loaded(store); + assert.equal((afterRace.head.todos as JsonObject[]).find(todo => todo.todo_id === target.todo_id)!.status, "open"); + }); for (const command of ["complete", "supersede"] as const) { test(`${provider}: ${command} replaces its own title without admitting duplicate work (${schema})`, async t => { const {store, contender} = await factory(t); diff --git a/tests/control_plane_ts/turn_envelope.test.ts b/tests/control_plane_ts/turn_envelope.test.ts index c73907b0b5..b85d97a003 100644 --- a/tests/control_plane_ts/turn_envelope.test.ts +++ b/tests/control_plane_ts/turn_envelope.test.ts @@ -14,6 +14,32 @@ import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; import { TURN_ENVELOPE_SECTION_TARGETS } from "../../loopx/control_plane/quota/turn_envelope_budget.ts"; import { projectPeerOrchestration } from "../../loopx/control_plane/quota/peer_orchestration.ts"; +test("task text commitment has one selected-Todo carrier without changing legacy context elision", () => { + for (const retainDetail of [false, true]) { + const source = payload(); + const channel = (source.interaction_contract as JsonObject).agent_channel as JsonObject; + channel.work_context = {complete: true, selected_todo_ref: "selected_todo", + instruction: "Read full task requirements.", selected_todo_text_verified: true, + ...(retainDetail ? {sources: [{source: "selected_todo", content: {todo: {note: "Retain evidence."}}}]} : {})}; + const envelope = buildTurnEnvelope({payload: source, + protocol_action_fields: protocolActionFields, scheduler_execution_args: ""}); + const selected = (envelope.action as JsonObject).selected_todo as JsonObject; + assert.equal(selected.text_source_verified, true); + assert.equal(envelope.work_context === undefined, !retainDetail); + assert.equal((envelope.work_context as JsonObject | undefined)?.selected_todo_text_verified, undefined); + assert.equal(JSON.stringify(envelope).split('"text_source_verified"').length - 1, 1); + assert.equal((envelope.action_signature as JsonObject).matches, true); + const legacy = structuredClone(source); + delete (((legacy.interaction_contract as JsonObject).agent_channel as JsonObject).work_context as JsonObject).selected_todo_text_verified; + const oldEnvelope = buildTurnEnvelope({payload: legacy, + protocol_action_fields: protocolActionFields, scheduler_execution_args: ""}); + assert.equal(((oldEnvelope.action as JsonObject).selected_todo as JsonObject).text_source_verified, undefined); + assert.equal((oldEnvelope.action_signature as JsonObject).matches, true); + assert.notEqual((oldEnvelope.action_signature as JsonObject).source_hash, + (envelope.action_signature as JsonObject).source_hash); + } +}); + function payload(): Record { return { ok: true, diff --git a/tests/test_loopx_turn_journal_inspection.py b/tests/test_loopx_turn_journal_inspection.py index 7d77653359..e1e0faea7e 100644 --- a/tests/test_loopx_turn_journal_inspection.py +++ b/tests/test_loopx_turn_journal_inspection.py @@ -612,8 +612,9 @@ def unexpected_call(*args: object, **kwargs: object) -> None: "scheduler_execution_context_for_turn", ): monkeypatch.setattr(turn_decision, name, unexpected_call) - for name in ("complete_goal_todo", "update_goal_todo"): - monkeypatch.setattr(turn_todo_writeback, name, unexpected_call) + # Patch the actual Turn write adapter; the retired update_goal_todo alias + # no longer exists here and cannot act as an inspection safety oracle. + monkeypatch.setattr(turn_todo_writeback, "complete_goal_todo", unexpected_call) monkeypatch.setattr(executor, "execute_turn_driver_settlement", unexpected_call) monkeypatch.setattr(executor, "_write_journal", unexpected_call)