diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md index 8268919d36..272820c937 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -535,10 +535,13 @@ Unknown command types fail closed. Transfer or delegated assignment, arbitrary todo/gate mutation, quota reservation, and external effects still require later runtime contracts and qualification. Non-empty write scopes and cross-todo scope-overlap rejection likewise require a later command contract -and qualification. The recoverable-execution verbs below are the Stage 3 -slice; steps 1 through 4 and 7 through 10 of Section 5 (identity, digest, -replay, CAS, reload, rebase, budget) apply to every verb unchanged, and only -the per-verb preconditions and transition (steps 5 and 6) differ. +and qualification. The recoverable-execution verbs below were called Stage 3 +in the historical #3669 implementation sequence. Under the current delivery +sequence in Section 11, that merged work is part of the Stage 0 reference foundation, +not the Stage 3 remote-shadow phase in Section 11. Steps 1 through +4 and 7 through 10 of Section 5 (identity, digest, replay, CAS, reload, rebase, +budget) apply to every verb unchanged, and only the per-verb preconditions and +transition (steps 5 and 6) differ. ### 5.2 `renew_work` @@ -1079,6 +1082,41 @@ The sequence is: cache, offline projection, and diagnostic material. Never keep a long-lived dual-write or dual-master mode. +#### Stage 2C observation foundation: local post-commit capture + +The first half of Stage 2C is an explicit, default-off product path. Preview +and enable it with: + +```bash +loopx configure-goal --goal-id GOAL --local-authority-shadow-file +loopx configure-goal --goal-id GOAL --local-authority-shadow-file --execute +``` + +Todo, handoff-mode, follow-up, and task-lease facades sample the full current +local projection after their primary write returns committed, then ask +`FileAuthorityStore` to retain that snapshot. `observation_trigger` records +why sampling began; it is not the primary transaction identity. A concurrent +primary commit may therefore appear in the sampled snapshot. A `captured` or +`replayed` result proves only the candidate-side observation commit. It does +not compare the source and candidate and carries `parity_verdict=not_evaluated`. + +Candidate bytes live under +`authority-shadow/file/` outside the legacy per-Goal runtime tree, so state +migration never copies a store identity or revision; an executed migration +seeds a new target lineage from the migrated local state. Candidate failure is +reported as an observation result but never reverses the completed local write. + +Disable the observer in one command with +`loopx configure-goal --goal-id GOAL --clear-local-authority-shadow --execute`. +This is rollback of observation only: the local Markdown and task-lease files +remain canonical throughout. The slice does not read the candidate for a +decision, fence a legacy writer, qualify a remote provider, or complete the +second Stage 2C promotion. A process crash after the local commit but before +the observer call may miss that individual observation; a later committed +write or migration seed refreshes the full current projection, but no durable +shadow outbox or transaction-correlated receipt is claimed here. This plumbing +is not parity evidence and cannot by itself support Stage 2C promotion. + ### Implementation prerequisite: put local file mode behind the same coordination contract Before wiring a live NoKV or another remote provider, the runtime should first diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md index 40e1348e31..40d8f25cb2 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md @@ -461,9 +461,11 @@ todo 已提交就要求调用方重新发一条 operation。 未知 command type fail closed。transfer 或 delegated assignment、任意 todo/gate mutation、quota reservation 与 external effect 仍需要后续 runtime 合同与 qualification；非空 write scope 与跨 todo scope-overlap 拒绝同样需要后续 command -contract 与 qualification。下面的可恢复执行动词是 Stage 3 切片；第 5 节的步骤 -1-4 与 7-10（identity、digest、replay、CAS、reload、rebase、budget）对每个动词 -原样适用，只有每动词的前置条件与迁移（步骤 5-6）不同。 +contract 与 qualification。下面的可恢复执行动词在 #3669 历史实施序列中曾称为 +Stage 3；按照第 11 节当前的交付编号，已合入的这部分属于 Stage 0 reference foundation， +不是第 11 节的 Stage 3 远端 shadow 阶段。第 5 节的步骤 1-4 与 7-10 +（identity、digest、replay、CAS、reload、rebase、budget）对每个动词原样适用，只有 +每动词的前置条件与迁移（步骤 5-6）不同。 ### 5.2 `renew_work` @@ -867,6 +869,36 @@ Stage 3/4 qualification 必须保持以下 ownership 与 proof 边界： LoopX service 成为唯一 writer。本地 `.loopx` 退为 cache、offline projection 与 诊断材料。绝不长期维持 dual-write 或 dual-master。 +#### Stage 2C 观察基础：本地提交后 capture + +Stage 2C 的前半段是一个显式开启、默认关闭的产品路径。先预览，再开启： + +```bash +loopx configure-goal --goal-id GOAL --local-authority-shadow-file +loopx configure-goal --goal-id GOAL --local-authority-shadow-file --execute +``` + +Todo、handoff-mode、follow-up 与 task-lease facade 会在本地主写返回成功后，采样 +完整当前本地投影，再让 `FileAuthorityStore` 保存该 snapshot。 +`observation_trigger` 只记录为何开始采样，不是主写 transaction identity；并发主写 +因此可能出现在该次 snapshot 中。`captured` 或 `replayed` 只证明候选侧 observation +commit，不表示已经对比 source 与 candidate；结果明确携带 +`parity_verdict=not_evaluated`。 + +候选数据位于 legacy 单 Goal runtime tree +之外的 `authority-shadow/file/`，因此 state migration 不会复制 store identity 或 +revision；真正执行迁移时，会从迁移后的本地主状态为目标端建立一条新 lineage。 +候选失败只形成 observation result，不会推翻已经完成的本地写入。 + +用 +`loopx configure-goal --goal-id GOAL --clear-local-authority-shadow --execute` +即可关闭 observer。这里回退的只是观察路径：Markdown 与 task-lease 文件始终是 +canonical。本切片不会读取候选来决策，不会 fence legacy writer，不会资格化远端 +provider，也没有完成 Stage 2C 后半段的本地 canonical promotion。若进程恰好在本地 +提交后、observer 调用前崩溃，该次 observation 可能丢失；后续成功写入或 migration +seed 会刷新完整当前投影，但这里不宣称已有 durable shadow outbox 或与主写 transaction +关联的 receipt。这套 plumbing 不是 parity evidence，不能单独支持 Stage 2C promotion。 + ### 实施前置条件：先让本地文件模式经过同一协调合同 在接入 live NoKV 或其他远端 provider 之前，runtime 应先把当前 todo/lease 写路径中的 diff --git a/loopx/cli.py b/loopx/cli.py index 04fd726696..99ed4caca4 100644 --- a/loopx/cli.py +++ b/loopx/cli.py @@ -807,6 +807,7 @@ def main(argv: list[str] | None = None) -> int: registry_path=registry_path, output_format=output_format, print_payload=print_payload, + runtime_root_arg=args.runtime_root, ) if handoff_mode_result is not None: return handoff_mode_result diff --git a/loopx/cli_commands/handoff_mode.py b/loopx/cli_commands/handoff_mode.py index ace7b7d739..d6becd214a 100644 --- a/loopx/cli_commands/handoff_mode.py +++ b/loopx/cli_commands/handoff_mode.py @@ -101,6 +101,7 @@ def handle_handoff_mode_command( registry_path: Path, output_format: Callable[..., str], print_payload: PrintPayload, + runtime_root_arg: str | None = None, ) -> int | None: if args.command != "handoff-mode": return None @@ -124,6 +125,7 @@ def handle_handoff_mode_command( registry_path=registry_path, goal_id=args.goal_id, mode=args.mode, + runtime_root_arg=runtime_root_arg, **path_args, ) except HandoffModeError as exc: diff --git a/loopx/cli_commands/registry_admin.py b/loopx/cli_commands/registry_admin.py index 1a8d96490c..2419449eb3 100644 --- a/loopx/cli_commands/registry_admin.py +++ b/loopx/cli_commands/registry_admin.py @@ -494,6 +494,12 @@ def handle_registry_admin_command( write_scope=args.write_scope, replace_write_scope=bool(args.replace_write_scope), clear_write_scope=bool(args.clear_write_scope), + local_authority_shadow_file=bool( + args.local_authority_shadow_file + ), + clear_local_authority_shadow=bool( + args.clear_local_authority_shadow + ), waiting_on=args.waiting_on, clear_waiting_on=bool(args.clear_waiting_on), boundary_authority_scopes=args.boundary_authority_scope, diff --git a/loopx/cli_commands/registry_admin_configure.py b/loopx/cli_commands/registry_admin_configure.py index c9b904920d..a174c8e52c 100644 --- a/loopx/cli_commands/registry_admin_configure.py +++ b/loopx/cli_commands/registry_admin_configure.py @@ -251,6 +251,22 @@ def register_configure_goal_command(subparsers: argparse._SubParsersAction) -> N action="store_true", help="Clear coordination.write_scope.", ) + configure_goal_parser.add_argument( + "--local-authority-shadow-file", + action="store_true", + help=( + "Enable default-off, one-way capture of post-commit local snapshots " + "in FileAuthorityStore. This does not compare source and candidate." + ), + ) + configure_goal_parser.add_argument( + "--clear-local-authority-shadow", + action="store_true", + help=( + "Disable the local authority shadow. This does not delete retained " + "candidate observations." + ), + ) configure_goal_parser.add_argument( "--waiting-on", choices=["codex", "user_or_controller", "controller", "external_evidence"], diff --git a/loopx/cli_commands/todo.py b/loopx/cli_commands/todo.py index d000360e61..a00bf02e75 100644 --- a/loopx/cli_commands/todo.py +++ b/loopx/cli_commands/todo.py @@ -779,6 +779,7 @@ def handle_todo_command( ) payload = add_goal_todo( registry_path=registry_path, + runtime_root_arg=runtime_root_arg, goal_id=args.goal_id, role=args.role, text=args.text, @@ -839,6 +840,7 @@ def handle_todo_command( validate_todo_claim_options(args) payload = update_goal_todo( registry_path=registry_path, + runtime_root_arg=runtime_root_arg, goal_id=args.goal_id, todo_id=args.todo_id, role=args.role, @@ -852,6 +854,7 @@ def handle_todo_command( validate_todo_update_options(args) payload = update_goal_todo( registry_path=registry_path, + runtime_root_arg=runtime_root_arg, goal_id=args.goal_id, todo_id=args.todo_id, text=args.text, @@ -993,6 +996,7 @@ def handle_todo_command( if completion_error is None: payload = complete_goal_todo( registry_path=registry_path, + runtime_root_arg=runtime_root_arg, goal_id=args.goal_id, todo_id=args.todo_id, role=args.role, @@ -1032,6 +1036,7 @@ def handle_todo_command( validate_todo_supersede_options(args) payload = supersede_goal_todo( registry_path=registry_path, + runtime_root_arg=runtime_root_arg, goal_id=args.goal_id, todo_id=args.todo_id, role=args.role, @@ -1057,6 +1062,7 @@ def handle_todo_command( validate_todo_archive_completed_options(args) payload = archive_completed_todos( registry_path=registry_path, + runtime_root_arg=runtime_root_arg, goal_id=args.goal_id, role=args.role or "agent", max_active_done=args.max_active_done, @@ -1081,6 +1087,7 @@ def handle_todo_command( followups.append(args.text) payload = capture_followup_todos( registry_path=registry_path, + runtime_root_arg=runtime_root_arg, goal_id=args.goal_id, followups=followups, evidence=args.evidence or "", diff --git a/loopx/configuration_catalog.py b/loopx/configuration_catalog.py index 11de69072f..051a935040 100644 --- a/loopx/configuration_catalog.py +++ b/loopx/configuration_catalog.py @@ -77,6 +77,11 @@ def build_goal_configuration_catalog( if isinstance(feature_summary.get("peer_task_coordination"), Mapping) else {} ) + local_authority_shadow = ( + feature_summary.get("local_authority_shadow") + if isinstance(feature_summary.get("local_authority_shadow"), Mapping) + else {} + ) graph_enable_args = ("--explore-graph-enabled",) harness_enable_args = ( "--explore-harness-enabled", @@ -100,6 +105,55 @@ def build_goal_configuration_catalog( ), }, "features": [ + { + "feature_id": "local_authority_shadow", + "display_name": "Local post-commit authority observation", + "availability": "experimental_opt_in", + "default": {"enabled": False}, + "current": { + "enabled": local_authority_shadow.get("enabled") is True, + "mode": local_authority_shadow.get("mode"), + "status": local_authority_shadow.get("status", "disabled"), + }, + "consider_when": ( + "A Goal needs to exercise the first Stage 2C observation " + "plumbing while legacy local writers remain authoritative." + ), + "effect": ( + "Captures a best-effort post-commit snapshot of Todo and " + "task-lease state through the FileAuthorityStore contract." + ), + "does_not": [ + "read the candidate for lifecycle decisions", + "write candidate state back into Markdown or task-lease files", + "promote shared authority or fence legacy writers", + "bind the snapshot to the exact primary transaction", + "guarantee delivery through a durable outbox", + "compare source and candidate or issue a parity verdict", + ], + "commands": { + "preview_enable": _configure_command( + goal_id, "--local-authority-shadow-file" + ), + "apply_enable": _configure_command( + goal_id, "--local-authority-shadow-file", execute=True + ), + "preview_disable": _configure_command( + goal_id, "--clear-local-authority-shadow" + ), + "apply_disable": _configure_command( + goal_id, "--clear-local-authority-shadow", execute=True + ), + "verify": [inspect_command], + }, + "documentation": { + "path": "docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md", + "url": ( + "https://github.com/huangruiteng/loopx/blob/main/" + "docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md" + ), + }, + }, { "feature_id": "multi_subagent", "display_name": "Adaptive child capacity", diff --git a/loopx/configure_goal.py b/loopx/configure_goal.py index d73987af3c..f71c2c7b62 100644 --- a/loopx/configure_goal.py +++ b/loopx/configure_goal.py @@ -34,6 +34,8 @@ ) from .control_plane.agents.supervisor import normalize_peer_supervisor from .control_plane.agents.work_mode import normalize_agent_work_modes +from .control_plane.coordination import local_authority_shadow_adapter as shadow +from .control_plane.coordination.configuration import normalize_goal_write_scope from .control_plane.operator_inbox_binding import local_private_config_digest from .control_plane.reward_memory import ( reward_memory_goal_policy, @@ -221,18 +223,6 @@ def _clean_registered_agents(values: list[str] | None) -> list[str] | None: return agents -def _clean_write_scope(values: list[str] | None) -> list[str] | None: - if values is None: - return None - scopes: list[str] = [] - for value in values: - for part in str(value).split(","): - scope = part.strip() - if scope and scope not in scopes: - scopes.append(scope) - return scopes - - def _settings_summary(goal: dict[str, Any]) -> dict[str, Any]: quota = goal_quota_config(goal) control_plane = compact_control_plane_policy(goal.get("control_plane")) @@ -259,7 +249,9 @@ def _settings_summary(goal: dict[str, Any]) -> dict[str, Any]: "explore_graph": compact_explore_graph_policy(goal.get("explore_graph")), "orchestration": orchestration, "waiting_on": goal.get("waiting_on"), - "write_scope": _clean_write_scope(coordination.get("write_scope") or []) or [], + "write_scope": normalize_goal_write_scope(coordination.get("write_scope") or []) + or [], + "local_authority_shadow": shadow.local_authority_shadow_summary(goal), "checkpointed_boundary_authority": checkpointed_boundary_authority_summary( coordination ), @@ -461,6 +453,8 @@ def configure_goal( write_scope: list[str] | None = None, replace_write_scope: bool = False, clear_write_scope: bool = False, + local_authority_shadow_file: bool = False, + clear_local_authority_shadow: bool = False, waiting_on: str | None = None, clear_waiting_on: bool = False, boundary_authority_scopes: list[str] | None = None, @@ -535,6 +529,9 @@ def configure_goal( raise ValueError( "--clear-write-scope cannot be combined with --replace-write-scope" ) + shadow.validate_local_authority_shadow_change( + local_authority_shadow_file, clear_local_authority_shadow + ) if clear_waiting_on and waiting_on: raise ValueError("--clear-waiting-on cannot be combined with --waiting-on") adding_boundary_authority = any( @@ -634,7 +631,7 @@ def configure_goal( clear_todo_lifecycle_authority ) supervised_agents = _clean_registered_agents(supervised_agents) - write_scope = _clean_write_scope(write_scope) + write_scope = normalize_goal_write_scope(write_scope) issue_fix_reviewer_notification_config = _local_private_config_path( issue_fix_reviewer_notification_config, label="reviewer notification config", @@ -1218,10 +1215,12 @@ def configure_goal( coordination["write_scope"] = write_scope else: existing_write_scope = ( - _clean_write_scope(coordination.get("write_scope") or []) or [] + normalize_goal_write_scope(coordination.get("write_scope") or []) + or [] ) coordination["write_scope"] = ( - _clean_write_scope([*existing_write_scope, *write_scope]) or [] + normalize_goal_write_scope([*existing_write_scope, *write_scope]) + or [] ) if clear_boundary_authority: coordination.pop("checkpointed_boundary_authority", None) @@ -1241,6 +1240,9 @@ def configure_goal( coordination["checkpointed_boundary_authority"] = [*entries, entry] goal["coordination"] = coordination + shadow.apply_local_authority_shadow_change( + goal, local_authority_shadow_file, clear_local_authority_shadow + ) after = _settings_summary(goal) changed_fields = _changed_fields(before, after) if goal != before_goal and not changed_fields: @@ -1279,6 +1281,7 @@ def configure_goal( "peer_task_coordination": deepcopy( after.get("peer_task_coordination") or {"enabled": False} ), + "local_authority_shadow": deepcopy(after["local_authority_shadow"]), "lark_event_inbox": _lark_event_inbox_config_summary(goal), "lark_kanban_heartbeat_sync": _lark_kanban_heartbeat_config_summary(goal), "reward_memory": reward_memory_goal_policy_summary(goal), diff --git a/loopx/control_plane/coordination/configuration.py b/loopx/control_plane/coordination/configuration.py new file mode 100644 index 0000000000..5a5e814966 --- /dev/null +++ b/loopx/control_plane/coordination/configuration.py @@ -0,0 +1,25 @@ +"""Normalization rules for goal-level coordination configuration.""" + +from __future__ import annotations + + +def normalize_goal_write_scope(values: list[str] | None) -> list[str] | None: + """Split comma-delimited goal scopes and preserve first-seen order. + + Goal configuration intentionally accepts a wider token vocabulary than a + Todo's ``required_write_scopes`` contract, so its stricter normalizer is + not interchangeable with this rule. + """ + + if values is None: + return None + scopes: list[str] = [] + for value in values: + for part in str(value).split(","): + scope = part.strip() + if scope and scope not in scopes: + scopes.append(scope) + return scopes + + +__all__ = ["normalize_goal_write_scope"] diff --git a/loopx/control_plane/coordination/local_authority_shadow.ts b/loopx/control_plane/coordination/local_authority_shadow.ts new file mode 100644 index 0000000000..b455cf5e04 --- /dev/null +++ b/loopx/control_plane/coordination/local_authority_shadow.ts @@ -0,0 +1,326 @@ +import { join } from "node:path"; + +import type { JsonObject } from "../effect_program.ts"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts"; +import type { + AuthorityStore, + AuthorityStoreLoadResult, + AuthorityStoreReceiptResult, +} from "./authority_store.ts"; +import { FileAuthorityStore } from "./file_authority_store.ts"; + +export const LOCAL_AUTHORITY_SHADOW_REQUEST_SCHEMA = + "loopx_local_authority_shadow_request_v0"; +export const LOCAL_AUTHORITY_SHADOW_PROJECTION_SCHEMA = + "loopx_local_authority_shadow_projection_v0"; +export const LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA = + "loopx_local_authority_shadow_evidence_v0"; +export const LOCAL_AUTHORITY_SHADOW_OBSERVATION_RECEIPT_SCHEMA = + "loopx_local_authority_shadow_observation_receipt_v0"; + +const REQUEST_FIELDS = new Set([ + "schema_version", + "mode", + "runtime_root", + "goal_id", + "observation_id", + "observation_trigger", + "source_digest", + "source_projection", +]); +export type LocalAuthorityShadowOutcome = + | "captured" + | "replayed" + | "ambiguous_reconciled" + | "ambiguous_unproved" + | "unavailable" + | "failed" + | "protocol_mismatch" + | "conflict_retry_required"; + +export interface LocalAuthorityShadowEvidence extends JsonObject { + schema_version: typeof LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA; + outcome: LocalAuthorityShadowOutcome; + reason_code: string | null; + goal_id: string; + observation_id: string; + source_digest: string; + capture_kind: "post_commit_snapshot"; + source_transaction_correlated: false; + durable_source_outbox: false; + source_candidate_compared: false; + parity_verdict: "not_evaluated"; + primary_authority: "legacy_local"; + candidate_provider: "file"; + candidate_read_for_decision: false; + provider_to_local_writes: false; + primary_writeback_preserved: true; + store_identity: string | null; + provider_revision: string | null; + cursor: string | null; +} + +interface LocalAuthorityShadowRequest { + mode: "file_one_way"; + runtime_root: string; + goal_id: string; + observation_id: string; + observation_trigger: string; + source_digest: string; + source_projection: JsonObject; +} + +export interface LocalAuthorityShadowDependencies { + openStore?: (directory: string, goalId: string) => AuthorityStore; +} + +function decodeRequest(value: unknown): LocalAuthorityShadowRequest { + const request = requireJsonObject(value, "local authority shadow request"); + const unexpected = Object.keys(request).filter((field) => !REQUEST_FIELDS.has(field)); + if (unexpected.length > 0) { + const listed = [...unexpected].sort((left, right) => (left < right ? -1 : left > right ? 1 : 0)); + throw new EffectRuntimeRequestError( + `Local authority shadow request has unsupported fields: ${listed.join(", ")}`, + ); + } + if (request.schema_version !== LOCAL_AUTHORITY_SHADOW_REQUEST_SCHEMA) { + throw new EffectRuntimeRequestError("Local authority shadow request schema mismatch"); + } + if (request.mode !== "file_one_way") { + throw new EffectRuntimeRequestError("Local authority shadow mode must be file_one_way"); + } + const goalId = requireNonEmptyString(request.goal_id, "goal_id"); + if (goalId === "." || goalId === ".." || goalId.includes("/") || goalId.includes("\\")) { + throw new EffectRuntimeRequestError( + "Local authority shadow goal id must be a single path segment", + ); + } + const projection = requireJsonObject(request.source_projection, "source_projection"); + if ( + projection.schema_version !== LOCAL_AUTHORITY_SHADOW_PROJECTION_SCHEMA || + projection.goal_id !== goalId + ) { + throw new EffectRuntimeRequestError( + "Local authority shadow projection schema or goal identity mismatch", + ); + } + const sourceDigest = requireNonEmptyString(request.source_digest, "source_digest"); + if (!/^sha256:[a-f0-9]{64}$/u.test(sourceDigest)) { + throw new EffectRuntimeRequestError("source_digest must be sha256:<64 lowercase hex>"); + } + return { + mode: "file_one_way", + runtime_root: requireNonEmptyString(request.runtime_root, "runtime_root"), + goal_id: goalId, + observation_id: requireNonEmptyString(request.observation_id, "observation_id"), + observation_trigger: requireNonEmptyString( + request.observation_trigger, + "observation_trigger", + ), + source_digest: sourceDigest, + source_projection: structuredClone(projection), + }; +} + +function evidence( + request: LocalAuthorityShadowRequest, + outcome: LocalAuthorityShadowOutcome, + options: { + reasonCode?: string | null; + storeIdentity?: string | null; + providerRevision?: string | null; + cursor?: string | null; + } = {}, +): LocalAuthorityShadowEvidence { + return { + schema_version: LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA, + outcome, + reason_code: options.reasonCode ?? null, + goal_id: request.goal_id, + observation_id: request.observation_id, + source_digest: request.source_digest, + capture_kind: "post_commit_snapshot", + source_transaction_correlated: false, + durable_source_outbox: false, + source_candidate_compared: false, + parity_verdict: "not_evaluated", + primary_authority: "legacy_local", + candidate_provider: "file", + candidate_read_for_decision: false, + provider_to_local_writes: false, + primary_writeback_preserved: true, + store_identity: options.storeIdentity ?? null, + provider_revision: options.providerRevision ?? null, + cursor: options.cursor ?? null, + }; +} + +function readFailureEvidence( + request: LocalAuthorityShadowRequest, + result: Extract, + storeIdentity: string | null, +): LocalAuthorityShadowEvidence { + return evidence(request, result.status, { + reasonCode: result.reason_code, + storeIdentity, + }); +} + +function receiptMatches( + request: LocalAuthorityShadowRequest, + result: Extract, +): boolean { + return result.receipts.some((raw) => { + const receipt = raw as Record; + return receipt.schema_version === LOCAL_AUTHORITY_SHADOW_OBSERVATION_RECEIPT_SCHEMA && + receipt.observation_id === request.observation_id && + receipt.source_digest === request.source_digest && + receipt.primary_authority === "legacy_local" && + receipt.provider_to_local_writes === false; + }); +} + +async function reconcileReceipt( + store: AuthorityStore, + request: LocalAuthorityShadowRequest, + storeIdentity: string, + reconciledOutcome: "replayed" | "ambiguous_reconciled", +): Promise { + const result = await store.readReceipt(request.observation_id); + if (result.status === "found" && receiptMatches(request, result)) { + return evidence(request, reconciledOutcome, { + storeIdentity, + providerRevision: result.provider_revision, + cursor: result.cursor, + }); + } + if (result.status === "unavailable") { + return evidence(request, "unavailable", { + reasonCode: result.reason_code, + storeIdentity, + }); + } + if (result.status === "failed") { + return evidence(request, "failed", { + reasonCode: result.reason_code, + storeIdentity, + }); + } + return evidence( + request, + reconciledOutcome === "ambiguous_reconciled" + ? "ambiguous_unproved" + : "protocol_mismatch", + { + reasonCode: result.status === "missing" + ? "observation_receipt_missing" + : "observation_receipt_mismatch", + storeIdentity, + }, + ); +} + +/** + * Record a post-commit observation in a candidate AuthorityStore. + * + * The legacy local writers remain the only decision authority. This function + * receives a completed source projection and has no route back to those files. + */ +export async function recordLocalAuthorityShadow( + value: unknown, + dependencies: LocalAuthorityShadowDependencies = {}, +): Promise { + const request = decodeRequest(value); + let store: AuthorityStore; + try { + const providerDirectory = join( + request.runtime_root, + "authority-shadow", + "file", + request.goal_id, + ); + store = (dependencies.openStore ?? ((directory, goalId) => + new FileAuthorityStore(directory, goalId)))( + providerDirectory, + request.goal_id, + ); + } catch { + return evidence(request, "unavailable", { + reasonCode: "provider_construction_failed", + }); + } + + try { + const identity = await store.storeIdentity(); + if (identity.status !== "available") { + return evidence(request, identity.status, { reasonCode: identity.reason_code }); + } + const storeIdentity = identity.store_identity; + const receipt = { + schema_version: LOCAL_AUTHORITY_SHADOW_OBSERVATION_RECEIPT_SCHEMA, + observation_id: request.observation_id, + source_digest: request.source_digest, + observation_trigger: request.observation_trigger, + source_transaction_correlated: false, + parity_verdict: "not_evaluated", + primary_authority: "legacy_local", + candidate_read_for_decision: false, + provider_to_local_writes: false, + }; + const event = { + schema_version: "loopx_local_authority_shadow_event_v0", + kind: "post_commit_snapshot_captured", + observation_id: request.observation_id, + observation_trigger: request.observation_trigger, + source_digest: request.source_digest, + }; + + const loaded = await store.loadAuthority(); + if (loaded.status === "unavailable" || loaded.status === "failed") { + return readFailureEvidence(request, loaded, storeIdentity); + } + const committed = await store.commitAuthority({ + expected_provider_revision: + loaded.status === "loaded" ? loaded.provider_revision : null, + operation_id: request.observation_id, + events: [event], + next_projection: request.source_projection, + receipts: [receipt], + }); + if (committed.status === "applied") { + return evidence(request, "captured", { + storeIdentity, + providerRevision: committed.provider_revision, + cursor: committed.cursor, + }); + } + if (committed.status === "ambiguous") { + return await reconcileReceipt( + store, + request, + storeIdentity, + "ambiguous_reconciled", + ); + } + if (committed.status === "failed") { + return evidence(request, "failed", { + reasonCode: committed.reason_code, + storeIdentity, + }); + } + if (committed.conflict_kind === "operation_id_exists") { + return await reconcileReceipt(store, request, storeIdentity, "replayed"); + } + return evidence(request, "conflict_retry_required", { + reasonCode: "provider_revision_mismatch", + storeIdentity, + providerRevision: committed.current_provider_revision, + cursor: committed.current_cursor, + }); + } catch { + return evidence(request, "unavailable", { + reasonCode: "provider_call_failed", + }); + } +} diff --git a/loopx/control_plane/coordination/local_authority_shadow_adapter.py b/loopx/control_plane/coordination/local_authority_shadow_adapter.py new file mode 100644 index 0000000000..fed81cb32f --- /dev/null +++ b/loopx/control_plane/coordination/local_authority_shadow_adapter.py @@ -0,0 +1,491 @@ +"""Post-commit bridge from legacy local authority into a file shadow. + +The adapter deliberately owns no lifecycle decision. It is entered only after +the existing Markdown or task-lease writer has succeeded, projects public-safe +facts, and asks the TypeScript authority-store boundary to retain an +observation. Missing configuration is a zero-effect fast path. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +from ...file_lock import LockAcquireTimeoutError, exclusive_file_lock +from ...history import load_registry +from ...paths import resolve_runtime_root +from ...registry import find_registry_goal +from ..effect_runtime import effect_runtime_result + + +LOCAL_AUTHORITY_SHADOW_CONFIG_SCHEMA = "loopx_local_authority_shadow_config_v0" +LOCAL_AUTHORITY_SHADOW_REQUEST_SCHEMA = "loopx_local_authority_shadow_request_v0" +LOCAL_AUTHORITY_SHADOW_PROJECTION_SCHEMA = "loopx_local_authority_shadow_projection_v0" +LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA = "loopx_local_authority_shadow_evidence_v0" +_CONFIG_FIELDS = {"schema_version", "mode"} +_PROJECTION_ATTEMPTS = 3 +_CONFLICT_RETRY_ATTEMPTS = 3 +_EVIDENCE_OUTCOMES = { + "captured", + "replayed", + "ambiguous_reconciled", + "ambiguous_unproved", + "unavailable", + "failed", + "protocol_mismatch", + "conflict_retry_required", +} +_TODO_FIELDS = ( + "todo_id", + "role", + "status", + "claimed_by", + "bound_agent", + "goal_bound", + "blocks_agent", + "excluded_agents", + "global_gate", + "task_class", + "action_kind", + "required_write_scopes", + "required_capabilities", + "continuation_policy", + "successor_todo_ids", + "no_followup", + "completion_continuation", +) +_LEASE_FIELDS = ( + "todo_id", + "owner", + "idempotency_key", + "write_scopes", + "version", + "lease_epoch", + "acquired_at", + "updated_at", + "expires_at", + "released_at", + "status", +) + + +def local_authority_shadow_summary(goal: Mapping[str, Any]) -> dict[str, Any]: + """Project the closed local-shadow configuration for operator readback.""" + + coordination = ( + goal.get("coordination") + if isinstance(goal.get("coordination"), Mapping) + else {} + ) + raw = coordination.get("authority_shadow") + if raw is None: + return {"enabled": False, "mode": None, "status": "disabled"} + valid = bool( + isinstance(raw, Mapping) + and set(raw) == _CONFIG_FIELDS + and raw.get("schema_version") == LOCAL_AUTHORITY_SHADOW_CONFIG_SCHEMA + and raw.get("mode") == "file_one_way" + ) + return { + "enabled": valid, + "mode": raw.get("mode") if isinstance(raw, Mapping) else None, + "status": "enabled" if valid else "invalid", + } + + +def validate_local_authority_shadow_change( + enable_file: bool, + clear: bool, +) -> None: + """Reject contradictory CLI intent before reading or mutating the registry.""" + + if enable_file and clear: + raise ValueError( + "--local-authority-shadow-file cannot be combined with " + "--clear-local-authority-shadow" + ) + + +def apply_local_authority_shadow_change( + goal: dict[str, Any], + enable_file: bool, + clear: bool, +) -> None: + """Apply a validated default-off local-shadow configuration change.""" + + if not enable_file and not clear: + return + coordination = ( + goal.get("coordination") if isinstance(goal.get("coordination"), dict) else {} + ) + if clear: + coordination.pop("authority_shadow", None) + else: + coordination["authority_shadow"] = { + "schema_version": LOCAL_AUTHORITY_SHADOW_CONFIG_SCHEMA, + "mode": "file_one_way", + } + if coordination: + goal["coordination"] = coordination + else: + goal.pop("coordination", None) + + +def effective_runtime_root( + registry_path: Path, + runtime_root_override: str | Path | None, +) -> Path: + """Resolve the one runtime root every writer hook of a CLI call must share. + + ``--runtime-root`` wins when given; otherwise the registry's + ``common_runtime_root`` applies, and a relative value resolves against the + registry's project root rather than the caller's working directory. Todo, + follow-up, handoff-mode, and task-lease hooks all consume this value so one + goal never splits into two candidate lineages. + """ + + registry = load_registry(registry_path) + override = str(runtime_root_override) if runtime_root_override is not None else None + return resolve_runtime_root(registry, override, registry_path=registry_path) + + +def _base_evidence( + *, + goal_id: str, + outcome: str, + reason_code: str, +) -> dict[str, Any]: + return { + "schema_version": LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA, + "outcome": outcome, + "reason_code": reason_code, + "goal_id": goal_id, + "observation_id": None, + "source_digest": None, + "capture_kind": "post_commit_snapshot", + "source_transaction_correlated": False, + "durable_source_outbox": False, + "source_candidate_compared": False, + "parity_verdict": "not_evaluated", + "primary_authority": "legacy_local", + "candidate_provider": "file", + "candidate_read_for_decision": False, + "provider_to_local_writes": False, + "primary_writeback_preserved": True, + "store_identity": None, + "provider_revision": None, + "cursor": None, + } + + +def _shadow_config(registry: dict[str, Any], goal_id: str) -> dict[str, str] | None: + goal = find_registry_goal(registry, goal_id) + coordination = goal.get("coordination") if isinstance(goal, dict) else None + if not isinstance(coordination, dict) or "authority_shadow" not in coordination: + return None + raw = coordination.get("authority_shadow") + if ( + not isinstance(raw, dict) + or set(raw) != _CONFIG_FIELDS + or raw.get("schema_version") != LOCAL_AUTHORITY_SHADOW_CONFIG_SCHEMA + or raw.get("mode") != "file_one_way" + ): + raise ValueError("authority_shadow must be a closed file_one_way config") + return {"mode": "file_one_way"} + + +def _canonical(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def _compact_todo(raw: object) -> dict[str, Any] | None: + if not isinstance(raw, dict): + return None + todo_id = str(raw.get("todo_id") or "").strip() + if not todo_id: + return None + compact = {field: raw[field] for field in _TODO_FIELDS if field in raw} + compact["todo_id"] = todo_id + if "status" not in compact and isinstance(raw.get("done"), bool): + compact["status"] = "done" if raw["done"] else "open" + return json.loads(_canonical(compact)) + + +def _compact_lease(path: Path, *, goal_id: str) -> dict[str, Any]: + raw = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise ValueError("task lease must contain an object") + if raw.get("goal_id") != goal_id or raw.get("todo_id") != path.stem: + raise ValueError("task lease identity does not match its shadow source") + return json.loads( + _canonical({field: raw[field] for field in _LEASE_FIELDS if field in raw}) + ) + + +def _source_projection( + *, + registry_path: Path, + runtime_root: Path, + goal_id: str, +) -> dict[str, Any]: + from ...control_plane.todos.handoff_mode import goal_handoff_mode_for_goal + from ...todos import list_goal_todos + + todo_payload = list_goal_todos( + registry_path=registry_path, + goal_id=goal_id, + runtime_root_arg=str(runtime_root), + ) + todos = [ + compact + for raw in todo_payload.get("todos") or [] + if (compact := _compact_todo(raw)) is not None + ] + todos.sort(key=lambda item: str(item["todo_id"])) + lease_dir = runtime_root / "goals" / goal_id / "task-leases" + leases = ( + [ + _compact_lease(path, goal_id=goal_id) + for path in sorted(lease_dir.glob("*.json")) + ] + if lease_dir.exists() + else [] + ) + projection = { + "schema_version": LOCAL_AUTHORITY_SHADOW_PROJECTION_SCHEMA, + "goal_id": goal_id, + "handoff_mode": goal_handoff_mode_for_goal( + registry_path=registry_path, + goal_id=goal_id, + ), + "todos": todos, + "leases": leases, + } + return json.loads(_canonical(projection)) + + +def _stable_projection( + *, + registry_path: Path, + runtime_root: Path, + goal_id: str, +) -> dict[str, Any]: + previous = _source_projection( + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=goal_id, + ) + for _attempt in range(_PROJECTION_ATTEMPTS): + current = _source_projection( + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=goal_id, + ) + if current == previous: + return current + previous = current + raise RuntimeError("local authority sources did not stabilize for shadowing") + + +def _valid_evidence( + result: object, + *, + goal_id: str, + observation_id: str, + source_digest: str, +) -> bool: + if not isinstance(result, dict): + return False + return ( + result.get("schema_version") == LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA + and result.get("outcome") in _EVIDENCE_OUTCOMES + and result.get("goal_id") == goal_id + and result.get("observation_id") == observation_id + and result.get("source_digest") == source_digest + and result.get("capture_kind") == "post_commit_snapshot" + and result.get("source_transaction_correlated") is False + and result.get("durable_source_outbox") is False + and result.get("source_candidate_compared") is False + and result.get("parity_verdict") == "not_evaluated" + and result.get("primary_authority") == "legacy_local" + and result.get("candidate_provider") == "file" + and result.get("candidate_read_for_decision") is False + and result.get("provider_to_local_writes") is False + and result.get("primary_writeback_preserved") is True + and ( + result.get("reason_code") is None + or isinstance(result.get("reason_code"), str) + ) + ) + + +def observe_local_authority_commit( + *, + registry_path: Path, + runtime_root: Path | None, + goal_id: str, + observation_trigger: str, +) -> dict[str, Any] | None: + """Capture a best-effort post-commit snapshot without changing its verdict. + + ``observation_trigger`` is diagnostic context, not a primary transaction + identity. The snapshot may include commits that landed after that trigger. + """ + + if not goal_id or goal_id in {".", ".."} or "/" in goal_id or "\\" in goal_id: + return _base_evidence( + goal_id=goal_id, + outcome="failed", + reason_code="invalid_shadow_goal_id", + ) + try: + registry = load_registry(registry_path) + config = _shadow_config(registry, goal_id) + except Exception: + return _base_evidence( + goal_id=goal_id, + outcome="failed", + reason_code="invalid_shadow_config", + ) + if config is None: + return None + + try: + if runtime_root is None: + runtime_root = resolve_runtime_root( + registry, + None, + registry_path=registry_path, + ) + # Candidate-provider bytes live outside the legacy per-goal runtime + # tree. State migration may copy that tree, but it must never copy a + # store identity or revision and accidentally create a second lineage. + shadow_root = runtime_root / "authority-shadow" / "file" / goal_id + with exclusive_file_lock( + shadow_root / "observation", + timeout_seconds=1.0, + operation="local_authority_shadow_observe", + ): + result: dict[str, Any] | None = None + for _attempt in range(_CONFLICT_RETRY_ATTEMPTS): + projection = _stable_projection( + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=goal_id, + ) + source_digest = ( + "sha256:" + hashlib.sha256(_canonical(projection)).hexdigest() + ) + observation_id = ( + "local-shadow:" + + hashlib.sha256( + _canonical( + { + "goal_id": goal_id, + "observation_trigger": observation_trigger, + "source_digest": source_digest, + } + ) + ).hexdigest() + ) + raw_result = effect_runtime_result( + "coordination.local_authority_shadow.record", + { + "schema_version": LOCAL_AUTHORITY_SHADOW_REQUEST_SCHEMA, + "mode": config["mode"], + "runtime_root": str(runtime_root), + "goal_id": goal_id, + "observation_id": observation_id, + "observation_trigger": observation_trigger, + "source_digest": source_digest, + "source_projection": projection, + }, + timeout=15.0, + ) + if not _valid_evidence( + raw_result, + goal_id=goal_id, + observation_id=observation_id, + source_digest=source_digest, + ): + return _base_evidence( + goal_id=goal_id, + outcome="failed", + reason_code="shadow_observation_result_invalid", + ) + result = dict(raw_result) + if result["outcome"] != "conflict_retry_required": + return result + if result is not None: + return result + except LockAcquireTimeoutError: + return _base_evidence( + goal_id=goal_id, + outcome="unavailable", + reason_code="shadow_observation_lock_timeout", + ) + except Exception: + return _base_evidence( + goal_id=goal_id, + outcome="failed", + reason_code="shadow_observation_failed", + ) + return _base_evidence( + goal_id=goal_id, + outcome="failed", + reason_code="shadow_observation_failed", + ) + + +def observe_todo_local_authority_commit( + payload: dict[str, Any], + registry_path: Path, + goal_id: str, + write_class: str, + *, + runtime_root: Path | None = None, +) -> dict[str, Any]: + """Attach post-commit shadow evidence without changing the Todo verdict. + + ``runtime_root`` is the effective root the writer resolved for this call; + ``None`` falls back to the registry root exactly as the other hooks do. + """ + + changed = any( + payload.get(field) + for field in ("changed", "added", "metadata_updated", "completed", "superseded") + ) + if payload.get("dry_run") or not changed: + return payload + todo_id = str(payload.get("todo_id") or "none") + updated_at = str(payload.get("updated_at") or "unknown") + evidence = observe_local_authority_commit( + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=goal_id, + observation_trigger=f"{write_class}:{todo_id}:{updated_at}", + ) + if evidence is not None: + payload["authority_shadow"] = evidence + return payload + + +__all__ = [ + "LOCAL_AUTHORITY_SHADOW_CONFIG_SCHEMA", + "LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA", + "apply_local_authority_shadow_change", + "effective_runtime_root", + "local_authority_shadow_summary", + "observe_local_authority_commit", + "observe_todo_local_authority_commit", + "validate_local_authority_shadow_change", +] diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 0d4dc8c651..23e432aff5 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -97,6 +97,7 @@ import { executeTaskLeaseAcquire, } from "./work_items/task_lease_acquire.ts"; import { executeTaskLeaseLifecycle } from "./work_items/task_lease_lifecycle.ts"; +import { recordLocalAuthorityShadow } from "./coordination/local_authority_shadow.ts"; import { evaluateTaskLeaseLifecycleDecision } from "./work_items/task_lease_lifecycle_decision.ts"; import { bootstrapCoordinationRuntimeShadow, @@ -396,6 +397,7 @@ export function createEffectRuntimeHandlers( ], ["task_lease.write_scopes.overlap", evaluateTaskLeaseWriteScopesOverlap], ["quota.monitor_poll.commit", evaluateQuotaMonitorPollCommit], + ["coordination.local_authority_shadow.record", recordLocalAuthorityShadow], [ "effect.program_from_ordered_steps", (params) => effectProgramFromOrderedSteps( diff --git a/loopx/control_plane/todos/goal_todo_projection.py b/loopx/control_plane/todos/goal_todo_projection.py new file mode 100644 index 0000000000..532315fa47 --- /dev/null +++ b/loopx/control_plane/todos/goal_todo_projection.py @@ -0,0 +1,345 @@ +"""Project todo summaries and items from one active-state text. + +The text is a parameter rather than a file read so a writer that still holds +the state-file lock can project the exact bytes it is about to commit; +``loopx.todos.list_goal_todos`` passes the on-disk text. Everything here is a +deterministic function of the text, the goal record, and the event projection. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from ...status import active_state_event_projection_fields +from .active_state_editing import TODO_SECTION_HEADINGS +from .active_state_todo_parser import parse_active_state_todos +from .list_projection import compact_explicit_limit_todo_summary +from .contract import ( + build_todo_id, + normalize_todo_blocks_agent, + normalize_todo_bound_agent, + normalize_todo_claimed_by, + normalize_todo_excluded_agents, + normalize_todo_id, + normalize_todo_status, +) +from .todo_summary import compact_todo_group, todo_item_status + + +def empty_todo_summary(*, role: str) -> dict[str, Any]: + return { + "schema_version": "todo_summary_v0", + "role": role, + "source_section": TODO_SECTION_HEADINGS[role], + "total_count": 0, + "open_count": 0, + "done_count": 0, + "items": [], + "first_open_items": [], + } + +def _user_todo_visible_to_agent(item: dict[str, Any], agent_id: str) -> bool: + if bool(item.get("global_gate")): + return True + blocks_agent = normalize_todo_blocks_agent(item.get("blocks_agent")) + if blocks_agent: + return blocks_agent == agent_id + bound_agent = normalize_todo_bound_agent(item.get("bound_agent")) + if bound_agent: + return bound_agent == agent_id + return True + +def filtered_todo_summary( + summary: dict[str, Any] | None, + *, + role: str, + status: str | None = None, + todo_id: str | None = None, + agent_id: str | None = None, + resume_source_items: list[dict[str, Any]] | None = None, + rollout_events: list[dict[str, Any]] | None = None, + item_limit: int | None = None, +) -> dict[str, Any]: + items = list((summary or {}).get("items") or []) + normalized_status = normalize_todo_status(status) + if normalized_status: + items = [item for item in items if todo_item_status(item) == normalized_status] + normalized_todo_id = normalize_todo_id(todo_id) if todo_id else None + if normalized_todo_id: + items = [ + item + for item in items + if normalize_todo_id(item.get("todo_id")) == normalized_todo_id + ] + normalized_agent_id = normalize_todo_claimed_by(agent_id) if agent_id else None + if normalized_agent_id: + if role == "agent": + items = [ + item + for item in items + if normalized_agent_id + not in normalize_todo_excluded_agents(item.get("excluded_agents")) + and ( + not normalize_todo_claimed_by(item.get("claimed_by")) + or normalize_todo_claimed_by(item.get("claimed_by")) + == normalized_agent_id + ) + ] + elif role == "user": + items = [ + item + for item in items + if _user_todo_visible_to_agent(item, normalized_agent_id) + ] + source_section = str((summary or {}).get("source_section") or TODO_SECTION_HEADINGS[role]) + return ( + compact_todo_group( + items, + source_section=source_section, + role=role, + resume_source_items=resume_source_items, + rollout_events=rollout_events, + item_limit=item_limit, + ) + or empty_todo_summary(role=role) + ) + +def summary_items(fields: dict[str, Any], role: str) -> list[dict[str, Any]]: + summary = fields.get(f"{role}_todos") if isinstance(fields, dict) else None + if not isinstance(summary, dict): + return [] + return [item for item in summary.get("items") or [] if isinstance(item, dict)] + +def merge_todo_projection_fields( + *, + markdown_fields: dict[str, Any], + event_fields: dict[str, Any], +) -> tuple[dict[str, Any], dict[str, Any]]: + merged: dict[str, Any] = {} + merged_items: dict[str, list[dict[str, Any]]] = {"user": [], "agent": []} + source_sections: dict[str, str] = {} + overlay: dict[str, Any] = { + "schema_version": "todo_list_projection_overlay_v0", + "base": "markdown_active_state", + "overlay": "event_projection", + "markdown_only_todo_ids": [], + "event_only_todo_ids": [], + "overlaid_todo_ids": [], + } + + # A todo_id is goal-wide identity. Merge both sources before splitting by + # role so an event-projected role change replaces the stale Markdown item. + by_id: dict[str, dict[str, Any]] = {} + order: list[str] = [] + markdown_ids: set[str] = set() + markdown_ids_by_role: dict[str, set[str]] = {"user": set(), "agent": set()} + event_ids: set[str] = set() + event_order: list[str] = [] + for role in ("user", "agent"): + markdown_items = summary_items(markdown_fields, role) + for item in markdown_items: + todo_id = normalize_todo_id(item.get("todo_id")) or build_todo_id( + role=role, + source_section=item.get("source_section"), + index=item.get("index"), + text=item.get("text"), + ) + if todo_id not in by_id: + order.append(todo_id) + markdown_ids.add(todo_id) + markdown_ids_by_role[role].add(todo_id) + by_id[todo_id] = dict(item) + + for role in ("user", "agent"): + event_items = summary_items(event_fields, role) + for item in event_items: + todo_id = normalize_todo_id(item.get("todo_id")) or build_todo_id( + role=role, + source_section=item.get("source_section"), + index=item.get("index"), + text=item.get("text"), + ) + if todo_id not in by_id: + order.append(todo_id) + if todo_id not in event_ids: + event_order.append(todo_id) + event_ids.add(todo_id) + by_id[todo_id] = dict(item) + + markdown_only_todo_ids: list[str] = [] + seen_markdown_only_ids: set[str] = set() + for role in ("user", "agent"): + for todo_id in sorted(markdown_ids_by_role[role] - event_ids): + if todo_id not in seen_markdown_only_ids: + markdown_only_todo_ids.append(todo_id) + seen_markdown_only_ids.add(todo_id) + overlay["markdown_only_todo_ids"] = markdown_only_todo_ids + overlay["event_only_todo_ids"] = [ + todo_id for todo_id in event_order if todo_id not in markdown_ids + ] + overlay["overlaid_todo_ids"] = [ + todo_id for todo_id in event_order if todo_id in markdown_ids + ] + + for todo_id in order: + item = by_id[todo_id] + final_role = "user" if item.get("role") == "user" else "agent" + merged_items[final_role].append(item) + + for role in ("user", "agent"): + source_section = str( + (markdown_fields.get(f"{role}_todos") or {}).get("source_section") + or (event_fields.get(f"{role}_todos") or {}).get("source_section") + or TODO_SECTION_HEADINGS[role] + ) + source_sections[role] = source_section + + resume_source_items = [*merged_items["user"], *merged_items["agent"]] + for role in ("user", "agent"): + if not merged_items[role]: + continue + summary = compact_todo_group( + merged_items[role], + source_section=source_sections[role], + role=role, + resume_source_items=resume_source_items, + item_limit=None, + ) + if summary: + merged[f"{role}_todos"] = summary + return merged, overlay + +class GoalTodoSummaries: + """Role summaries and todo items projected from one active-state text.""" + + __slots__ = ( + "source", + "projection_fields", + "projection_overlay", + "summaries", + "todos", + "unfiltered_count", + "uncapped_todo_count", + ) + + def __init__( + self, + *, + source: str, + projection_fields: dict[str, Any], + projection_overlay: dict[str, Any] | None, + summaries: dict[str, dict[str, Any]], + todos: list[dict[str, Any]], + unfiltered_count: int, + uncapped_todo_count: int, + ) -> None: + self.source = source + self.projection_fields = projection_fields + self.projection_overlay = projection_overlay + self.summaries = summaries + self.todos = todos + self.unfiltered_count = unfiltered_count + self.uncapped_todo_count = uncapped_todo_count + +def goal_todo_summaries( + goal: dict[str, Any] | None, + *, + state_text: str, + state_path: Path, + rollout_events: list[dict[str, Any]], + roles: list[str], + status: str | None, + todo_id: str | None, + agent_id: str | None, + limit: int | None, +) -> GoalTodoSummaries: + """Project todo summaries from active-state text plus its event projection. + + The text is a parameter rather than a file read so a writer that still + holds the state-file lock can project the exact bytes it is about to + commit; ``list_goal_todos`` passes the on-disk text. + """ + + projection_fields = active_state_event_projection_fields( + goal or {}, + state_path=state_path, + item_limit=None, + rollout_events=rollout_events, + ) + projection_has_todos = bool( + projection_fields.get("user_todos") or projection_fields.get("agent_todos") + ) + markdown_fields = parse_active_state_todos( + state_text, + goal=goal, + state_path=state_path, + item_limit=None, + rollout_events=rollout_events, + ) + markdown_has_todos = bool( + markdown_fields.get("user_todos") or markdown_fields.get("agent_todos") + ) + projection_overlay: dict[str, Any] | None = None + if projection_has_todos and markdown_has_todos: + fields, projection_overlay = merge_todo_projection_fields( + markdown_fields=markdown_fields, + event_fields=projection_fields, + ) + source = "event_projection_with_markdown_overlay" + elif projection_has_todos: + fields = projection_fields + source = "event_projection" + else: + fields = markdown_fields + source = "markdown_active_state" + + resume_source_items = [ + *summary_items(fields, "user"), + *summary_items(fields, "agent"), + ] + summaries: dict[str, dict[str, Any]] = {} + todos: list[dict[str, Any]] = [] + unfiltered_count = 0 + uncapped_todo_count = 0 + for item_role in roles: + key = f"{item_role}_todos" + raw_summary = fields.get(key) if isinstance(fields, dict) else None + unfiltered_count += len((raw_summary or {}).get("items") or []) + summary = filtered_todo_summary( + raw_summary, + role=item_role, + status=status, + todo_id=todo_id, + agent_id=agent_id, + resume_source_items=resume_source_items, + rollout_events=rollout_events, + item_limit=limit, + ) + if limit is not None: + summary = compact_explicit_limit_todo_summary( + summary, + role=item_role, + item_limit=limit, + ) + summaries[key] = summary + todos.extend(summary.get("items") or []) + uncapped_todo_count += int(summary.get("total_count") or 0) + return GoalTodoSummaries( + source=source, + projection_fields=projection_fields, + projection_overlay=projection_overlay, + summaries=summaries, + todos=todos, + unfiltered_count=unfiltered_count, + uncapped_todo_count=uncapped_todo_count, + ) + +__all__ = [ + "GoalTodoSummaries", + "empty_todo_summary", + "filtered_todo_summary", + "goal_todo_summaries", + "merge_todo_projection_fields", + "summary_items", +] diff --git a/loopx/control_plane/todos/handoff_mode.py b/loopx/control_plane/todos/handoff_mode.py index f0129b188f..80e0af5d46 100644 --- a/loopx/control_plane/todos/handoff_mode.py +++ b/loopx/control_plane/todos/handoff_mode.py @@ -138,6 +138,7 @@ def enter_todo_ownership_handoff_gate( mutation_authority: dict[str, Any], actor_agent_id: str | None, ownership_mutation: bool, + runtime_root: Path | None = None, ) -> dict[str, Any]: """Gate one claimed_by mutation on an existing todo behind the goal mode. @@ -168,6 +169,7 @@ def enter_todo_ownership_handoff_gate( goal_id=goal_id, todo_id=todo_id, actor_agent_id=actor_agent_id, + runtime_root=runtime_root, ) ) return extras @@ -184,6 +186,7 @@ def enter_added_todo_ownership_handoff_gate( text: str, claimed_by: str | None, actor_agent_id: str | None, + runtime_root: Path | None = None, ) -> dict[str, Any]: """Gate the ``todo add`` path when it would reassign an existing todo. @@ -225,6 +228,7 @@ def enter_added_todo_ownership_handoff_gate( and requested is not None and requested != current ), + runtime_root=runtime_root, ) @@ -294,6 +298,7 @@ def _quiescence_offenders( registry_path: Path, goal_id: str, state_text: str, + runtime_root: Path | None = None, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Return blockers visible to the v0 materialized-state scan. @@ -329,7 +334,8 @@ def _quiescence_offenders( } ) leases: list[dict[str, Any]] = [] - runtime_root = runtime_root_from_registry(registry_path, None) + if runtime_root is None: + runtime_root = runtime_root_from_registry(registry_path, None) lease_dir = task_lease_dir(runtime_root=runtime_root, goal_id=goal_id) if lease_dir.exists(): for path in sorted(lease_dir.glob("todo_*.json")): @@ -359,6 +365,34 @@ def _authority_offender_tokens( ) +def _previous_handoff_mode_fields( + previous_raw: object, +) -> tuple[str, dict[str, Any]]: + """Type the persisted front-matter mode; invalid values stay reportable.""" + + try: + previous = normalize_handoff_mode(previous_raw) + except HandoffModeError as exc: + previous = str(previous_raw or "").strip() + return previous, { + "previous_mode": previous, + "previous_mode_valid": False, + "previous_mode_error_code": exc.code, + } + return previous, {"previous_mode": previous, "previous_mode_valid": True} + + +def _write_handoff_mode_frontmatter(lines: list[str], requested: str) -> None: + """Replace or insert the handoff_mode key inside the front-matter block.""" + + open_index, close_index = _frontmatter_bounds(lines) + for index in range(open_index, close_index): + if lines[index].split(":", 1)[0].strip() == HANDOFF_MODE_FRONTMATTER_KEY: + lines[index] = f"{HANDOFF_MODE_FRONTMATTER_KEY}: {requested}" + return + lines.insert(close_index, f"{HANDOFF_MODE_FRONTMATTER_KEY}: {requested}") + + def set_goal_handoff_mode( *, registry_path: Path, @@ -366,6 +400,7 @@ def set_goal_handoff_mode( mode: str, project: Path | None = None, state_file: Path | None = None, + runtime_root_arg: str | None = None, ) -> dict[str, Any]: """Set the goal handoff mode; requires a quiescent goal for transitions. @@ -399,25 +434,14 @@ def set_goal_handoff_mode( project=project, state_file=state_file, ) + # One effective runtime root for the lease lock, the quiescence scan, and + # the post-commit observation of this call. + runtime_root = runtime_root_from_registry(registry_path, runtime_root_arg) with exclusive_file_lock(resolved_state_file, operation="handoff_mode_set"): original = resolved_state_file.read_text(encoding="utf-8") - previous_raw = parse_state_frontmatter(original).get( - HANDOFF_MODE_FRONTMATTER_KEY + previous, previous_mode_fields = _previous_handoff_mode_fields( + parse_state_frontmatter(original).get(HANDOFF_MODE_FRONTMATTER_KEY) ) - try: - previous = normalize_handoff_mode(previous_raw) - except HandoffModeError as exc: - previous = str(previous_raw or "").strip() - previous_mode_fields = { - "previous_mode": previous, - "previous_mode_valid": False, - "previous_mode_error_code": exc.code, - } - else: - previous_mode_fields = { - "previous_mode": previous, - "previous_mode_valid": True, - } payload = { "ok": True, "schema_version": HANDOFF_MODE_SCHEMA_VERSION, @@ -430,15 +454,13 @@ def set_goal_handoff_mode( if previous == requested: payload["changed"] = False return payload - lease_lock = task_lease_lock_path( - runtime_root=runtime_root_from_registry(registry_path, None), - goal_id=goal_id, - ) + lease_lock = task_lease_lock_path(runtime_root=runtime_root, goal_id=goal_id) with exclusive_file_lock(lease_lock, operation="handoff_mode_set"): claimed, leases = _quiescence_offenders( registry_path=registry_path, goal_id=goal_id, state_text=original, + runtime_root=runtime_root, ) requested_core_mode = HandoffMode(requested) if previous in HANDOFF_MODE_VALUES: @@ -494,19 +516,20 @@ def set_goal_handoff_mode( }, ) lines = original.splitlines() - open_index, close_index = _frontmatter_bounds(lines) - for index in range(open_index, close_index): - if ( - lines[index].split(":", 1)[0].strip() - == HANDOFF_MODE_FRONTMATTER_KEY - ): - lines[index] = f"{HANDOFF_MODE_FRONTMATTER_KEY}: {requested}" - break - else: - lines.insert( - close_index, f"{HANDOFF_MODE_FRONTMATTER_KEY}: {requested}" - ) + _write_handoff_mode_frontmatter(lines, requested) new_text = "\n".join(lines) + ("\n" if original.endswith("\n") else "") resolved_state_file.write_text(new_text, encoding="utf-8") payload["changed"] = True + from ..coordination.local_authority_shadow_adapter import ( + observe_local_authority_commit, + ) + + evidence = observe_local_authority_commit( + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=goal_id, + observation_trigger=f"handoff_mode_set:{previous}:{requested}", + ) + if evidence is not None: + payload["authority_shadow"] = evidence return payload diff --git a/loopx/control_plane/work_items/task_lease.py b/loopx/control_plane/work_items/task_lease.py index ebcdf5e9d8..f46d97eea2 100644 --- a/loopx/control_plane/work_items/task_lease.py +++ b/loopx/control_plane/work_items/task_lease.py @@ -262,6 +262,7 @@ def hold_handoff_lease_holder_gate( goal_id: str, todo_id: str, actor_agent_id: str | None, + runtime_root: Path | None = None, ) -> Iterator[dict[str, Any]]: """Hold the per-goal lease lock while proving the actor owns the lease. @@ -289,7 +290,8 @@ def hold_handoff_lease_holder_gate( "reason": "missing_actor", }, ) - runtime_root = runtime_root_from_registry(registry_path, None) + if runtime_root is None: + runtime_root = runtime_root_from_registry(registry_path, None) result = _execute_native_task_lease_lifecycle( runtime_root=runtime_root, registry_path=registry_path, @@ -362,6 +364,7 @@ def hold_task_lease_mutation_fence( expected_version: int | None = None, require_active_when_key_supplied: bool = True, handoff: dict[str, Any] | None = None, + runtime_root: Path | None = None, ) -> Iterator[dict[str, Any]]: """Hold the per-goal lease lock while one todo lifecycle write commits. @@ -384,7 +387,8 @@ def hold_task_lease_mutation_fence( normalized_goal_id = normalize_goal_id(goal_id) normalized_todo_id = normalize_lease_todo_id(todo_id) - runtime_root = runtime_root_from_registry(registry_path, None) + if runtime_root is None: + runtime_root = runtime_root_from_registry(registry_path, None) handoff = handoff or {} result = _execute_native_task_lease_lifecycle( runtime_root=runtime_root, @@ -474,6 +478,7 @@ def enter_terminal_todo_lease_fence( mutation_authority: dict[str, Any], idempotency_key: str | None = None, expected_version: int | None = None, + runtime_root: Path | None = None, ) -> tuple[dict[str, Any], dict[str, Any]]: """Resolve the goal handoff and hold the lease fence for one terminal write. @@ -502,6 +507,7 @@ def enter_terminal_todo_lease_fence( idempotency_key is not None or expected_version is not None ), handoff=handoff, + runtime_root=runtime_root, ) ) return handoff, fence @@ -552,8 +558,19 @@ def runtime_root_from_registry( registry_path: Path, runtime_root_override: str | None, ) -> Path: + """Effective runtime root for lease state: override, else registry root. + + A relative ``common_runtime_root`` resolves against the registry's project + root, matching the observation hooks, so the lease files and every + candidate observation of one goal share a single root. + """ + registry = load_registry(registry_path) - return resolve_runtime_root(registry, runtime_root_override) + return resolve_runtime_root( + registry, + runtime_root_override, + registry_path=registry_path, + ) def task_lease_todo_projection( diff --git a/loopx/control_plane/work_items/task_lease_acquire_adapter.py b/loopx/control_plane/work_items/task_lease_acquire_adapter.py index e96e335958..e277030df8 100644 --- a/loopx/control_plane/work_items/task_lease_acquire_adapter.py +++ b/loopx/control_plane/work_items/task_lease_acquire_adapter.py @@ -24,6 +24,44 @@ TASK_LEASE_AUTHORITY_SNAPSHOT_ATTEMPTS = 3 +def _attach_local_authority_shadow( + result: dict[str, Any], + *, + registry_path: Path | None, + runtime_root: Path, + goal_id: str, + todo_id: str, + operation: str, +) -> dict[str, Any]: + """Observe a committed public lease mutation without changing its verdict.""" + + if registry_path is None: + return result + lease = result.get("lease") if isinstance(result.get("lease"), dict) else {} + observation_trigger = ":".join( + ( + f"task_lease_{operation}", + str(todo_id), + str(lease.get("version") or "none"), + str(lease.get("lease_epoch") or "none"), + str(lease.get("updated_at") or lease.get("released_at") or "unknown"), + ) + ) + from ..coordination.local_authority_shadow_adapter import ( + observe_local_authority_commit, + ) + + evidence = observe_local_authority_commit( + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=str(goal_id), + observation_trigger=observation_trigger, + ) + if evidence is not None: + result["authority_shadow"] = evidence + return result + + def _authority_source_receipt(source_id: str, path: Path) -> dict[str, Any]: resolved = path.expanduser().resolve(strict=False) try: @@ -280,6 +318,43 @@ def task_lease_acquire_authority_facts( ) +def _require_native_acquire_shape(payload: object) -> dict[str, Any]: + if ( + not isinstance(payload, dict) + or payload.get("schema_version") != TASK_LEASE_SCHEMA_VERSION + or payload.get("action") != "acquire" + or not isinstance(payload.get("ok"), bool) + ): + raise RuntimeError("native task-lease acquire result shape mismatch") + return payload + + +def _finalize_native_acquire_result( + payload: dict[str, Any], + *, + authority: dict[str, Any], + registry_path: Path, + runtime_root: Path, + goal_id: str, + todo_id: str, + legacy_provider_projection: bool, +) -> dict[str, Any]: + result = dict(payload) + if legacy_provider_projection and result.get("ok") is True: + result["handoff_mode"] = authority.get("handoff_mode") or HANDOFF_MODE_LEGACY + result.pop("settlement", None) + if result.get("ok") is True and result.get("acquired") is True: + result = _attach_local_authority_shadow( + result, + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=goal_id, + todo_id=todo_id, + operation="acquire", + ) + return result + + def execute_native_task_lease_acquire( *, registry_path: Path, @@ -315,30 +390,23 @@ def execute_native_task_lease_acquire( "expected_version": expected_version, "authority": authority, } - payload = effect_runtime_result( - "task_lease.acquire.native", - request, - timeout=15.0, + payload = _require_native_acquire_shape( + effect_runtime_result("task_lease.acquire.native", request, timeout=15.0) ) - if ( - not isinstance(payload, dict) - or payload.get("schema_version") != TASK_LEASE_SCHEMA_VERSION - or payload.get("action") != "acquire" - or not isinstance(payload.get("ok"), bool) - ): - raise RuntimeError("native task-lease acquire result shape mismatch") if ( payload.get("error_code") == "authority_source_changed" and attempt + 1 < TASK_LEASE_AUTHORITY_SNAPSHOT_ATTEMPTS ): continue - result = dict(payload) - if _legacy_provider_projection and result.get("ok") is True: - result["handoff_mode"] = ( - authority.get("handoff_mode") or HANDOFF_MODE_LEGACY - ) - result.pop("settlement", None) - return result + return _finalize_native_acquire_result( + payload, + authority=authority, + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=str(goal_id), + todo_id=str(todo_id), + legacy_provider_projection=_legacy_provider_projection, + ) raise RuntimeError("native task-lease acquire exhausted source-CAS retries") @@ -592,6 +660,24 @@ def execute_native_task_lease_lifecycle( result["handoff_mode"] = authority["handoff_mode"] if _legacy_provider_projection: result.pop("settlement", None) + committed_mutation = ( + normalized_operation == "renew" and result.get("renewed") is True + ) or ( + normalized_operation == "transfer" + and result.get("transferred") is True + ) or ( + normalized_operation == "release" + and result.get("released") is True + ) + if committed_mutation and result.get("idempotent") is not True: + result = _attach_local_authority_shadow( + result, + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=str(goal_id), + todo_id=str(todo_id), + operation=normalized_operation, + ) # lock_token is an internal bridge value. Callers that need a held # fence read it from the nested native payload before redacting it. return result diff --git a/loopx/state_migration.py b/loopx/state_migration.py index 59cf867058..ca8380d0cf 100644 --- a/loopx/state_migration.py +++ b/loopx/state_migration.py @@ -2,19 +2,29 @@ import copy import json -import os import shutil from datetime import datetime, timezone from pathlib import Path from typing import Any from .global_registry import write_json -from .paths import DEFAULT_RUNTIME_ROOT from .registry import registry_goals LEGACY_RUNTIME_ROOT = Path.home() / ".codex" / "goal-harness" LEGACY_GLOBAL_REGISTRY = LEGACY_RUNTIME_ROOT / "registry.global.json" +AUTHORITY_SHADOW_CONFIG_SCHEMA = "loopx_local_authority_shadow_config_v0" +MIGRATION_SHADOW_SEED_EVIDENCE_SCHEMA = "loopx_state_migration_shadow_seed_evidence_v0" +_SHADOW_EVIDENCE_OUTCOMES = { + "captured", + "replayed", + "ambiguous_reconciled", + "ambiguous_unproved", + "unavailable", + "failed", + "protocol_mismatch", + "conflict_retry_required", +} def now_local() -> str: @@ -131,7 +141,7 @@ def copy_active_state_files( for source_goal, target_goal in pairs: source_path = resolve_goal_state(source_goal.get("repo"), source_goal.get("state_file")) target_path = resolve_goal_state(target_goal.get("repo"), target_goal.get("state_file")) - row = { + row: dict[str, Any] = { "goal_id": target_goal.get("id"), "source": str(source_path) if source_path else None, "target": str(target_path) if target_path else None, @@ -171,7 +181,7 @@ def copy_runtime_goal_dirs( new_goal_id = goal_id_map.get(old_goal_id, old_goal_id) source_dir = legacy_runtime_root.expanduser() / "goals" / old_goal_id target_dir = target_runtime_root.expanduser() / "goals" / new_goal_id - row = { + row: dict[str, Any] = { "source": str(source_dir), "target": str(target_dir), "goal_id": new_goal_id, @@ -206,6 +216,104 @@ def copy_runtime_goal_dirs( return results +def _uses_file_authority_shadow(goal: dict[str, Any]) -> bool: + coordination = goal.get("coordination") + if not isinstance(coordination, dict): + return False + config = coordination.get("authority_shadow") + return ( + isinstance(config, dict) + and config.get("schema_version") == AUTHORITY_SHADOW_CONFIG_SCHEMA + and config.get("mode") == "file_one_way" + ) + + +def _shadow_seed_evidence( + *, + goal_id: str, + attempted: bool, + outcome: str, + reason_code: str | None = None, +) -> dict[str, Any]: + return { + "schema_version": MIGRATION_SHADOW_SEED_EVIDENCE_SCHEMA, + "goal_id": goal_id, + "attempted": attempted, + "outcome": outcome, + "reason_code": reason_code, + } + + +def _shadow_seed_result(*, goal_id: str, result: object) -> dict[str, Any]: + outcome = result.get("outcome") if isinstance(result, dict) else None + if outcome not in _SHADOW_EVIDENCE_OUTCOMES: + return _shadow_seed_evidence( + goal_id=goal_id, + attempted=True, + outcome="failed", + reason_code="post_migration_shadow_seed_invalid_evidence", + ) + + reason_code = ( + None + if outcome in {"captured", "replayed", "ambiguous_reconciled"} + else f"post_migration_shadow_seed_{outcome}" + ) + return _shadow_seed_evidence( + goal_id=goal_id, + attempted=True, + outcome=str(outcome), + reason_code=reason_code, + ) + + +def seed_migrated_authority_shadows( + *, + goals: list[dict[str, Any]], + target_registry_path: Path, + target_runtime_root: Path, + execute: bool, +) -> list[dict[str, Any]]: + """Plan or seed fresh candidate lineage from migrated local authority.""" + + results: list[dict[str, Any]] = [] + for goal in goals: + if not _uses_file_authority_shadow(goal): + continue + goal_id = str(goal.get("id") or "") + if not execute: + results.append( + _shadow_seed_evidence( + goal_id=goal_id, + attempted=False, + outcome="planned", + ) + ) + continue + try: + from .control_plane.coordination.local_authority_shadow_adapter import ( + observe_local_authority_commit, + ) + + result = observe_local_authority_commit( + registry_path=target_registry_path, + runtime_root=target_runtime_root, + goal_id=goal_id, + observation_trigger="state_migration_seed", + ) + results.append(_shadow_seed_result(goal_id=goal_id, result=result)) + except Exception: + results.append( + _shadow_seed_evidence( + goal_id=goal_id, + attempted=True, + outcome="failed", + reason_code="post_migration_shadow_seed_failed", + ) + ) + return results + + def migrate_legacy_state( *, legacy_registry_path: Path, @@ -282,6 +390,13 @@ def migrate_legacy_state( if execute: write_json(target_registry_path, target_payload) + authority_shadow_seeds = seed_migrated_authority_shadows( + goals=incoming_goals, + target_registry_path=target_registry_path, + target_runtime_root=target_runtime_root, + execute=execute, + ) + return { "ok": True, "schema_version": "loopx_state_migration_v0", @@ -299,6 +414,7 @@ def migrate_legacy_state( "project_registry_goal_count": len(target_payload.get("goals", [])), "active_state": active_state_results, "runtime_goals": runtime_results, + "authority_shadow_seeds": authority_shadow_seeds, } @@ -338,6 +454,16 @@ def render_state_migration_markdown(payload: dict[str, Any]) -> str: ) if row.get("skipped_reason"): lines.append(f" - skipped_reason: `{row.get('skipped_reason')}`") + authority_shadow_seeds = payload.get("authority_shadow_seeds") or [] + if authority_shadow_seeds: + lines.extend(["", "## Authority Shadow Seeds"]) + for row in authority_shadow_seeds: + lines.append( + f"- `{row.get('goal_id')}` outcome=`{row.get('outcome')}` " + f"attempted=`{row.get('attempted')}`" + ) + if row.get("reason_code"): + lines.append(f" - reason_code: `{row.get('reason_code')}`") global_sync = payload.get("global_sync") if isinstance(global_sync, dict): lines.extend(["", "## Global Sync"]) diff --git a/loopx/todo_followups.py b/loopx/todo_followups.py index 45024052ae..a3dd7910c1 100644 --- a/loopx/todo_followups.py +++ b/loopx/todo_followups.py @@ -66,6 +66,7 @@ def capture_followup_todos( project: Path | None = None, state_file: Path | None = None, dry_run: bool = False, + runtime_root_arg: str | None = None, ) -> dict[str, Any]: if not followups: raise ValueError("todo capture-followups requires at least one --follow-up") @@ -149,7 +150,7 @@ def capture_followup_todos( if not dry_run: resolved_state_file.write_text(new_text, encoding="utf-8") - return { + result = { "ok": True, "dry_run": dry_run, "changed": changed, @@ -166,3 +167,20 @@ def capture_followup_todos( "items": items, "updated_at": updated_at if changed else None, } + if changed and not dry_run: + from .control_plane.coordination.local_authority_shadow_adapter import ( + effective_runtime_root, + observe_local_authority_commit, + ) + + shadow = observe_local_authority_commit( + registry_path=registry_path, + runtime_root=effective_runtime_root(registry_path, runtime_root_arg), + goal_id=goal_id, + observation_trigger=( + f"todo_capture_followups:{recorded_count}:{updated_at}" + ), + ) + if shadow is not None: + result["authority_shadow"] = shadow + return result diff --git a/loopx/todos.py b/loopx/todos.py index 3c345b2fe3..08d5d5b35b 100644 --- a/loopx/todos.py +++ b/loopx/todos.py @@ -10,11 +10,7 @@ from .paths import resolve_runtime_root from .rollout_event_log import load_rollout_events, rollout_event_log_path from .state_refresh import now_local, resolve_goal_state -from .status import ( - MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE, - active_state_event_projection_fields, -) -from .control_plane.todos.active_state_todo_parser import parse_active_state_todos +from .status import MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE from .control_plane.todos.contract import ( TodoContinuationPolicy, TODO_STATUS_DEFERRED, @@ -90,16 +86,15 @@ AGENT_LANE_OVERLAY_FULL_DETAIL_COLD_PATH, EXPLICIT_LIMIT_OVERLAY_FULL_DETAIL_COLD_PATH, compact_agent_lane_todo_summary, - compact_explicit_limit_todo_summary, compact_thin_todo_list_payload, compact_todo_projection_overlay, todo_item_relations, todo_list_projection_contract, ) +from .control_plane.todos.goal_todo_projection import goal_todo_summaries from .control_plane.todos import monitor_metadata as todo_monitor_metadata from .control_plane.todos.external_wait_writeback import plan_todo_external_wait_update from .control_plane.todos.mutation_authority import authorize_todo_lifecycle_mutation, todo_update_authority_action -from .control_plane.todos.todo_summary import compact_todo_group, todo_item_status from .control_plane.todos.succession_warning import build_open_parent_successor_advisory from .control_plane.todos.todo_index import MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL from .control_plane.todos.text import ( @@ -126,6 +121,10 @@ enter_todo_ownership_handoff_gate, resolve_todo_completion_handoff, ) +from .control_plane.coordination.local_authority_shadow_adapter import ( + effective_runtime_root, + observe_todo_local_authority_commit as _shadow_todo, +) from .control_plane.work_items.task_lease import ( enter_terminal_todo_lease_fence, hold_task_lease_mutation_fence, @@ -175,194 +174,6 @@ def resolve_todo_state_path( return resolved_project, resolved_state_file -def empty_todo_summary(*, role: str) -> dict[str, Any]: - return { - "schema_version": "todo_summary_v0", - "role": role, - "source_section": TODO_SECTION_HEADINGS[role], - "total_count": 0, - "open_count": 0, - "done_count": 0, - "items": [], - "first_open_items": [], - } - - -def _user_todo_visible_to_agent(item: dict[str, Any], agent_id: str) -> bool: - if bool(item.get("global_gate")): - return True - blocks_agent = normalize_todo_blocks_agent(item.get("blocks_agent")) - if blocks_agent: - return blocks_agent == agent_id - bound_agent = normalize_todo_bound_agent(item.get("bound_agent")) - if bound_agent: - return bound_agent == agent_id - return True - - -def filtered_todo_summary( - summary: dict[str, Any] | None, - *, - role: str, - status: str | None = None, - todo_id: str | None = None, - agent_id: str | None = None, - resume_source_items: list[dict[str, Any]] | None = None, - rollout_events: list[dict[str, Any]] | None = None, - item_limit: int | None = None, -) -> dict[str, Any]: - items = list((summary or {}).get("items") or []) - normalized_status = normalize_todo_status(status) - if normalized_status: - items = [item for item in items if todo_item_status(item) == normalized_status] - normalized_todo_id = normalize_todo_id(todo_id) if todo_id else None - if normalized_todo_id: - items = [ - item - for item in items - if normalize_todo_id(item.get("todo_id")) == normalized_todo_id - ] - normalized_agent_id = normalize_todo_claimed_by(agent_id) if agent_id else None - if normalized_agent_id: - if role == "agent": - items = [ - item - for item in items - if normalized_agent_id - not in normalize_todo_excluded_agents(item.get("excluded_agents")) - and ( - not normalize_todo_claimed_by(item.get("claimed_by")) - or normalize_todo_claimed_by(item.get("claimed_by")) - == normalized_agent_id - ) - ] - elif role == "user": - items = [ - item - for item in items - if _user_todo_visible_to_agent(item, normalized_agent_id) - ] - source_section = str((summary or {}).get("source_section") or TODO_SECTION_HEADINGS[role]) - return ( - compact_todo_group( - items, - source_section=source_section, - role=role, - resume_source_items=resume_source_items, - rollout_events=rollout_events, - item_limit=item_limit, - ) - or empty_todo_summary(role=role) - ) - - -def _summary_items(fields: dict[str, Any], role: str) -> list[dict[str, Any]]: - summary = fields.get(f"{role}_todos") if isinstance(fields, dict) else None - if not isinstance(summary, dict): - return [] - return [item for item in summary.get("items") or [] if isinstance(item, dict)] - - -def _merge_todo_projection_fields( - *, - markdown_fields: dict[str, Any], - event_fields: dict[str, Any], -) -> tuple[dict[str, Any], dict[str, Any]]: - merged: dict[str, Any] = {} - merged_items: dict[str, list[dict[str, Any]]] = {"user": [], "agent": []} - source_sections: dict[str, str] = {} - overlay: dict[str, Any] = { - "schema_version": "todo_list_projection_overlay_v0", - "base": "markdown_active_state", - "overlay": "event_projection", - "markdown_only_todo_ids": [], - "event_only_todo_ids": [], - "overlaid_todo_ids": [], - } - - # A todo_id is goal-wide identity. Merge both sources before splitting by - # role so an event-projected role change replaces the stale Markdown item. - by_id: dict[str, dict[str, Any]] = {} - order: list[str] = [] - markdown_ids: set[str] = set() - markdown_ids_by_role: dict[str, set[str]] = {"user": set(), "agent": set()} - event_ids: set[str] = set() - event_order: list[str] = [] - for role in ("user", "agent"): - markdown_items = _summary_items(markdown_fields, role) - for item in markdown_items: - todo_id = normalize_todo_id(item.get("todo_id")) or build_todo_id( - role=role, - source_section=item.get("source_section"), - index=item.get("index"), - text=item.get("text"), - ) - if todo_id not in by_id: - order.append(todo_id) - markdown_ids.add(todo_id) - markdown_ids_by_role[role].add(todo_id) - by_id[todo_id] = dict(item) - - for role in ("user", "agent"): - event_items = _summary_items(event_fields, role) - for item in event_items: - todo_id = normalize_todo_id(item.get("todo_id")) or build_todo_id( - role=role, - source_section=item.get("source_section"), - index=item.get("index"), - text=item.get("text"), - ) - if todo_id not in by_id: - order.append(todo_id) - if todo_id not in event_ids: - event_order.append(todo_id) - event_ids.add(todo_id) - by_id[todo_id] = dict(item) - - markdown_only_todo_ids: list[str] = [] - seen_markdown_only_ids: set[str] = set() - for role in ("user", "agent"): - for todo_id in sorted(markdown_ids_by_role[role] - event_ids): - if todo_id not in seen_markdown_only_ids: - markdown_only_todo_ids.append(todo_id) - seen_markdown_only_ids.add(todo_id) - overlay["markdown_only_todo_ids"] = markdown_only_todo_ids - overlay["event_only_todo_ids"] = [ - todo_id for todo_id in event_order if todo_id not in markdown_ids - ] - overlay["overlaid_todo_ids"] = [ - todo_id for todo_id in event_order if todo_id in markdown_ids - ] - - for todo_id in order: - item = by_id[todo_id] - final_role = "user" if item.get("role") == "user" else "agent" - merged_items[final_role].append(item) - - for role in ("user", "agent"): - source_section = str( - (markdown_fields.get(f"{role}_todos") or {}).get("source_section") - or (event_fields.get(f"{role}_todos") or {}).get("source_section") - or TODO_SECTION_HEADINGS[role] - ) - source_sections[role] = source_section - - resume_source_items = [*merged_items["user"], *merged_items["agent"]] - for role in ("user", "agent"): - if not merged_items[role]: - continue - summary = compact_todo_group( - merged_items[role], - source_section=source_sections[role], - role=role, - resume_source_items=resume_source_items, - item_limit=None, - ) - if summary: - merged[f"{role}_todos"] = summary - return merged, overlay - - def list_goal_todos( *, registry_path: Path, @@ -403,71 +214,25 @@ def list_goal_todos( limit=MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL, ) - projection_fields = active_state_event_projection_fields( + roles = [role] if role else ["user", "agent"] + projected = goal_todo_summaries( goal, + state_text=resolved_state_file.read_text(encoding="utf-8"), state_path=resolved_state_file, - item_limit=None, - rollout_events=rollout_events, - ) - projection_has_todos = bool( - projection_fields.get("user_todos") or projection_fields.get("agent_todos") - ) - markdown_fields = parse_active_state_todos( - resolved_state_file.read_text(encoding="utf-8"), - goal=goal, - state_path=resolved_state_file, - item_limit=None, rollout_events=rollout_events, + roles=roles, + status=status, + todo_id=normalized_todo_id, + agent_id=normalized_agent_id, + limit=limit, ) - markdown_has_todos = bool( - markdown_fields.get("user_todos") or markdown_fields.get("agent_todos") - ) - projection_overlay: dict[str, Any] | None = None - if projection_has_todos and markdown_has_todos: - fields, projection_overlay = _merge_todo_projection_fields( - markdown_fields=markdown_fields, - event_fields=projection_fields, - ) - source = "event_projection_with_markdown_overlay" - elif projection_has_todos: - fields = projection_fields - source = "event_projection" - else: - fields = markdown_fields - source = "markdown_active_state" - - roles = [role] if role else ["user", "agent"] - resume_source_items = [ - *_summary_items(fields, "user"), - *_summary_items(fields, "agent"), - ] - summaries: dict[str, dict[str, Any]] = {} - todos: list[dict[str, Any]] = [] - unfiltered_count = 0 - uncapped_todo_count = 0 - for item_role in roles: - key = f"{item_role}_todos" - raw_summary = fields.get(key) if isinstance(fields, dict) else None - unfiltered_count += len((raw_summary or {}).get("items") or []) - summary = filtered_todo_summary( - raw_summary, - role=item_role, - status=status, - todo_id=normalized_todo_id, - agent_id=normalized_agent_id, - resume_source_items=resume_source_items, - rollout_events=rollout_events, - item_limit=limit, - ) - if limit is not None: - summary = compact_explicit_limit_todo_summary( - summary, - role=item_role, - item_limit=limit, - ) - summaries[key] = summary - todos.extend(summary.get("items") or []) - uncapped_todo_count += int(summary.get("total_count") or 0) + source = projected.source + projection_fields = projected.projection_fields + projection_overlay = projected.projection_overlay + summaries = projected.summaries + todos = projected.todos + unfiltered_count = projected.unfiltered_count + uncapped_todo_count = projected.uncapped_todo_count matched_todo_count = len(todos) agent_lane_hot_path = bool( @@ -873,6 +638,7 @@ def add_goal_todo( *, registry_path: Path, goal_id: str, + runtime_root_arg: str | None = None, role: str, text: str, status: str | None = None, @@ -908,6 +674,7 @@ def add_goal_todo( state_file: Path | None = None, dry_run: bool = False, ) -> dict[str, Any]: + shadow_runtime_root = effective_runtime_root(registry_path, runtime_root_arg) if role not in TODO_SECTION_HEADINGS: raise ValueError("todo role must be one of: user, agent") require_user_todo_task_class( @@ -1070,6 +837,7 @@ def add_goal_todo( text=todo_text, claimed_by=effective_claimed_by, actor_agent_id=effective_agent_id or effective_claimed_by, + runtime_root=shadow_runtime_root, ) add_result = add_todo_to_lines( lines, @@ -1162,12 +930,13 @@ def add_goal_todo( "updated_at": updated_at if changed else None, **handoff_gate, } - return _attach_todo_write_correctness_dry_run_packet( + payload = _attach_todo_write_correctness_dry_run_packet( payload, goal_id=goal_id, write_class="todo_add", state_text=original, ) + return _shadow_todo(payload, registry_path, goal_id, "todo_add", runtime_root=shadow_runtime_root) def resolve_todo_state( @@ -1191,6 +960,7 @@ def update_goal_todo( *, registry_path: Path, goal_id: str, + runtime_root_arg: str | None = None, todo_id: str, text: str | None = None, status: str | None = None, @@ -1231,6 +1001,7 @@ def update_goal_todo( state_file: Path | None = None, dry_run: bool = False, ) -> dict[str, Any]: + shadow_runtime_root = effective_runtime_root(registry_path, runtime_root_arg) if excluded_agents and clear_excluded_agents: raise ValueError( "todo update accepts either excluded_agents or clear_excluded_agents, not both" @@ -1368,6 +1139,7 @@ def update_goal_todo( mutation_authority=mutation_authority, actor_agent_id=effective_agent_id or effective_claimed_by, ownership_mutation=(claimed_by is not None or clear_claim) and target_role == "agent", + runtime_root=shadow_runtime_root, ) target_task_class = task_class or str(existing_block.get("task_class") or "") if target_role == "user" and claimed_by: @@ -1618,18 +1390,20 @@ def update_goal_todo( payload["external_wait_transition"] = external_wait_transition if monitor_poll_transition is not None: payload["monitor_poll_transition"] = monitor_poll_transition - return _attach_todo_write_correctness_dry_run_packet( + payload = _attach_todo_write_correctness_dry_run_packet( payload, goal_id=goal_id, write_class=write_class, state_text=original, ) + return _shadow_todo(payload, registry_path, goal_id, write_class, runtime_root=shadow_runtime_root) def complete_goal_todo( *, registry_path: Path, goal_id: str, + runtime_root_arg: str | None = None, todo_id: str, role: str | None = None, decision_outcome: str | None = None, @@ -1659,6 +1433,7 @@ def complete_goal_todo( state_file: Path | None = None, dry_run: bool = False, ) -> dict[str, Any]: + shadow_runtime_root = effective_runtime_root(registry_path, runtime_root_arg) if next_task_repository and not next_agent_todo: raise ValueError("--next-task-repository requires --next-agent-todo") if next_required_capabilities and not next_agent_todo: @@ -1782,6 +1557,7 @@ def complete_goal_todo( or task_lease_expected_version is not None ), handoff=completion_handoff, + runtime_root=shadow_runtime_root, ) ) completion_state = completion_transaction.get("completion_state") @@ -1852,7 +1628,10 @@ def complete_goal_todo( task_lease_fence, committed=bool(event_result.get("changed")) and not dry_run, ) - return event_result + write_class = "todo_complete_event_projection" + return _shadow_todo( + event_result, registry_path, goal_id, write_class, runtime_root=shadow_runtime_root + ) if not isinstance(completion_state, dict): raise RuntimeError( "TypeScript Todo completion transaction did not authorize a commit" @@ -2010,12 +1789,13 @@ def complete_goal_todo( if effective_decision_outcome: result["decision_outcome"] = effective_decision_outcome result["self_merged"] = effective_self_merged - return result + return _shadow_todo(result, registry_path, goal_id, "todo_complete", runtime_root=shadow_runtime_root) def supersede_goal_todo( *, registry_path: Path, goal_id: str, + runtime_root_arg: str | None = None, todo_id: str, role: str | None = None, reason: str | None = None, @@ -2035,6 +1815,7 @@ def supersede_goal_todo( state_file: Path | None = None, dry_run: bool = False, ) -> dict[str, Any]: + shadow_runtime_root = effective_runtime_root(registry_path, runtime_root_arg) if next_task_repository and not next_agent_todo: raise ValueError("--next-task-repository requires --next-agent-todo") if next_required_capabilities and not next_agent_todo: @@ -2075,6 +1856,7 @@ def supersede_goal_todo( lease_fence_stack, registry_path=registry_path, goal_id=goal_id, todo_id=todo_id, todo=authority_todo, actor_agent_id=agent_id, state_text=original, mutation_authority=mutation_authority, idempotency_key=task_lease_idempotency_key, expected_version=task_lease_expected_version, + runtime_root=shadow_runtime_root, ) effective_next_claimed_by = ( require_registered_agent_id(registry_path=registry_path, goal_id=goal_id, agent_id=next_claimed_by, field="next_claimed_by") @@ -2205,7 +1987,7 @@ def supersede_goal_todo( if changed and not dry_run: resolved_state_file.write_text(new_text, encoding="utf-8") release_verified_task_lease_fence(task_lease_fence, committed=changed and not dry_run) - return { + result = { "ok": True, "dry_run": dry_run, "superseded": True, @@ -2219,18 +2001,21 @@ def supersede_goal_todo( "project": str(resolved_project) if resolved_project else None, "updated_at": updated_at if changed else None, } + return _shadow_todo(result, registry_path, goal_id, "todo_supersede", runtime_root=shadow_runtime_root) def archive_completed_todos( *, registry_path: Path, goal_id: str, + runtime_root_arg: str | None = None, role: str = "agent", max_active_done: int = ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE, project: Path | None = None, state_file: Path | None = None, dry_run: bool = True, ) -> dict[str, Any]: + shadow_runtime_root = effective_runtime_root(registry_path, runtime_root_arg) if role not in TODO_SECTION_HEADINGS: raise ValueError("todo role must be one of: user, agent") if max_active_done < 0: @@ -2262,7 +2047,7 @@ def archive_completed_todos( if changed and not dry_run: resolved_state_file.write_text(new_text, encoding="utf-8") - return { + result = { "ok": True, "dry_run": dry_run, "goal_id": goal_id, @@ -2271,3 +2056,6 @@ def archive_completed_todos( "project": str(resolved_project) if resolved_project else None, "updated_at": updated_at if changed else None, } + return _shadow_todo( + result, registry_path, goal_id, "todo_archive_completed", runtime_root=shadow_runtime_root + ) diff --git a/tests/control_plane/test_coordination_configuration.py b/tests/control_plane/test_coordination_configuration.py new file mode 100644 index 0000000000..10c0ca4789 --- /dev/null +++ b/tests/control_plane/test_coordination_configuration.py @@ -0,0 +1,11 @@ +from __future__ import annotations + +from loopx.control_plane.coordination.configuration import normalize_goal_write_scope + + +def test_goal_write_scope_normalization_preserves_configuration_semantics() -> None: + assert normalize_goal_write_scope(None) is None + assert normalize_goal_write_scope([]) == [] + assert normalize_goal_write_scope( + [" docs/**, tests/** ", "docs/**", "src/**;generated/**", ""] + ) == ["docs/**", "tests/**", "src/**;generated/**"] diff --git a/tests/control_plane/test_local_authority_shadow_cli_e2e.py b/tests/control_plane/test_local_authority_shadow_cli_e2e.py new file mode 100644 index 0000000000..8ab7cccd0d --- /dev/null +++ b/tests/control_plane/test_local_authority_shadow_cli_e2e.py @@ -0,0 +1,469 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess +import sys +import time + +import pytest + +from loopx.file_lock import exclusive_file_lock + + +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def _workspace(tmp_path: Path, *, goal_id: str) -> tuple[Path, Path, Path]: + repo = tmp_path / goal_id + repo.mkdir() + state = repo / "ACTIVE_GOAL_STATE.md" + state.write_text( + "---\n" + f"goal_id: {goal_id}\n" + "handoff_mode: hard_lease\n" + "updated_at: 2026-09-02T00:00:00+00:00\n" + "---\n\n" + "## Agent Todo\n\n", + encoding="utf-8", + ) + runtime_root = tmp_path / f"{goal_id}-runtime" + registry = tmp_path / f"{goal_id}-registry.json" + registry.write_text( + json.dumps( + { + "common_runtime_root": str(runtime_root), + "goals": [ + { + "id": goal_id, + "status": "active", + "repo": str(repo), + "state_file": state.name, + "coordination": { + "agent_model": "peer_v1", + "registered_agents": ["agent-a", "agent-b"], + }, + } + ], + } + ), + encoding="utf-8", + ) + return registry, state, runtime_root + + +def _command(registry: Path, runtime_root: Path, *args: str) -> list[str]: + return [ + sys.executable, + "-m", + "loopx.cli", + "--registry", + str(registry), + "--runtime-root", + str(runtime_root), + "--format", + "json", + *args, + ] + + +def _env() -> dict[str, str]: + env = os.environ.copy() + env["PYTHONPATH"] = str(REPO_ROOT) + return env + + +def _cli(registry: Path, runtime_root: Path, *args: str) -> dict[str, object]: + completed = subprocess.run( + _command(registry, runtime_root, *args), + cwd=REPO_ROOT, + env=_env(), + check=True, + capture_output=True, + text=True, + timeout=30, + ) + return json.loads(completed.stdout) + + +def _store_document(runtime_root: Path, goal_id: str) -> tuple[Path, dict[str, object]]: + paths = list( + (runtime_root / "authority-shadow" / "file" / goal_id).glob( + "authority-store-*.json" + ) + ) + assert len(paths) == 1 + return paths[0], json.loads(paths[0].read_text(encoding="utf-8")) + + +def _add_todo( + registry: Path, + runtime_root: Path, + *, + goal_id: str, + text: str, +) -> dict[str, object]: + return _cli( + registry, + runtime_root, + "todo", + "add", + "--goal-id", + goal_id, + "--role", + "agent", + "--text", + text, + "--task-class", + "advancement_task", + ) + + +def test_product_cli_configure_capture_readback_disable_and_default_off_lifecycle_isolation( + tmp_path: Path, +) -> None: + goal_id = "shadow-cli-e2e" + registry, _state, runtime_root = _workspace(tmp_path, goal_id=goal_id) + + preview = _cli( + registry, + runtime_root, + "configure-goal", + "--goal-id", + goal_id, + "--local-authority-shadow-file", + ) + assert preview["dry_run"] is True + assert preview["written"] is False + + enabled = _cli( + registry, + runtime_root, + "configure-goal", + "--goal-id", + goal_id, + "--local-authority-shadow-file", + "--execute", + ) + assert enabled["written"] is True + + text = "Capture one post-commit observation through the product CLI." + observed = _add_todo( + registry, + runtime_root, + goal_id=goal_id, + text=text, + ) + assert observed["authority_shadow"]["outcome"] == "captured" # type: ignore[index] + assert observed["authority_shadow"]["parity_verdict"] == "not_evaluated" # type: ignore[index] + lease = _cli( + registry, + runtime_root, + "task-lease", + "acquire", + "--goal-id", + goal_id, + "--todo-id", + str(observed["todo_id"]), + "--owner", + "agent-a", + "--idempotency-key", + "shadow-cli-lease", + "--ttl-seconds", + "120", + ) + assert lease["acquired"] is True + assert lease["authority_shadow"]["outcome"] == "captured" # type: ignore[index] + store_path, store = _store_document(runtime_root, goal_id) + assert len(store["head"]["todos"]) == 1 # type: ignore[index] + assert len(store["head"]["leases"]) == 1 # type: ignore[index] + + inspected = _cli( + registry, + runtime_root, + "configure-goal", + "--goal-id", + goal_id, + ) + assert inspected["after"]["local_authority_shadow"] == { # type: ignore[index] + "enabled": True, + "mode": "file_one_way", + "status": "enabled", + } + + disabled = _cli( + registry, + runtime_root, + "configure-goal", + "--goal-id", + goal_id, + "--clear-local-authority-shadow", + "--execute", + ) + assert disabled["written"] is True + candidate_before_disabled_write = store_path.read_bytes() + after_disable = _add_todo( + registry, + runtime_root, + goal_id=goal_id, + text="This local lifecycle write must not execute the observer.", + ) + assert after_disable["ok"] is True + assert after_disable["added"] is True + assert "authority_shadow" not in after_disable + assert store_path.read_bytes() == candidate_before_disabled_write + + baseline_registry, _baseline_state, baseline_runtime = _workspace( + tmp_path, goal_id="shadow-cli-baseline" + ) + baseline = _add_todo( + baseline_registry, + baseline_runtime, + goal_id="shadow-cli-baseline", + text=text, + ) + for field in ( + "ok", + "added", + "already_exists", + "metadata_updated", + "status_changed", + "role", + "status", + "task_class", + "action_kind", + "continuation_policy", + ): + assert observed[field] == baseline[field] + assert not (baseline_runtime / "authority-shadow").exists() + + +def test_product_cli_candidate_failure_preserves_the_primary_lifecycle_commit( + tmp_path: Path, +) -> None: + goal_id = "shadow-cli-failure" + registry, state, runtime_root = _workspace(tmp_path, goal_id=goal_id) + _cli( + registry, + runtime_root, + "configure-goal", + "--goal-id", + goal_id, + "--local-authority-shadow-file", + "--execute", + ) + runtime_root.mkdir(parents=True, exist_ok=True) + (runtime_root / "authority-shadow").write_text("block candidate directory", encoding="utf-8") + + result = _add_todo( + registry, + runtime_root, + goal_id=goal_id, + text="The primary write survives a candidate construction failure.", + ) + + assert result["ok"] is True + assert result["added"] is True + assert result["authority_shadow"]["outcome"] == "failed" # type: ignore[index] + assert result["authority_shadow"]["reason_code"] == "shadow_observation_failed" # type: ignore[index] + assert str(result["todo_id"]) in state.read_text(encoding="utf-8") + + +@pytest.mark.skipif(os.name == "nt", reason="requires POSIX cross-process flock and SIGKILL") +def test_product_cli_loses_capture_between_commit_and_observer_then_refreshes_snapshot( + tmp_path: Path, +) -> None: + goal_id = "shadow-cli-crash-gap" + registry, state, runtime_root = _workspace(tmp_path, goal_id=goal_id) + _cli( + registry, + runtime_root, + "configure-goal", + "--goal-id", + goal_id, + "--local-authority-shadow-file", + "--execute", + ) + first_text = "Primary commit that loses its post-commit observation." + observation_lock_target = ( + runtime_root / "authority-shadow" / "file" / goal_id / "observation" + ) + + with exclusive_file_lock(observation_lock_target, operation="e2e_crash_gap"): + process = subprocess.Popen( + _command( + registry, + runtime_root, + "todo", + "add", + "--goal-id", + goal_id, + "--role", + "agent", + "--text", + first_text, + "--task-class", + "advancement_task", + ), + cwd=REPO_ROOT, + env=_env(), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + deadline = time.monotonic() + 5.0 + while first_text not in state.read_text(encoding="utf-8"): + if time.monotonic() >= deadline: + process.kill() + process.communicate(timeout=5) + raise AssertionError("primary Todo commit did not become visible") + time.sleep(0.01) + assert process.poll() is None + process.kill() + process.communicate(timeout=5) + + assert not list( + (runtime_root / "authority-shadow" / "file" / goal_id).glob( + "authority-store-*.json" + ) + ) + + recovered = _add_todo( + registry, + runtime_root, + goal_id=goal_id, + text="A later primary commit refreshes the current full snapshot.", + ) + assert recovered["authority_shadow"]["outcome"] == "captured" # type: ignore[index] + assert recovered["authority_shadow"]["durable_source_outbox"] is False # type: ignore[index] + assert recovered["authority_shadow"]["source_transaction_correlated"] is False # type: ignore[index] + assert recovered["authority_shadow"]["parity_verdict"] == "not_evaluated" # type: ignore[index] + _store_path, store = _store_document(runtime_root, goal_id) + assert len(store["head"]["todos"]) == 2 # type: ignore[index] + + +def test_product_cli_runtime_root_override_keeps_one_candidate_lineage( + tmp_path: Path, +) -> None: + """``--runtime-root`` differs from ``common_runtime_root``: one lineage, one head. + + Every writer family of one CLI invocation must observe into the same + candidate store: Todo add, task-lease acquire, Todo update, follow-up + capture, and a leased completion. The registry root must not gain a + candidate lineage of its own. + """ + + goal_id = "shadow-cli-one-root" + registry, state, registry_runtime = _workspace(tmp_path, goal_id=goal_id) + override_runtime = tmp_path / f"{goal_id}-override-runtime" + assert override_runtime != registry_runtime + _cli( + registry, + override_runtime, + "configure-goal", + "--goal-id", + goal_id, + "--local-authority-shadow-file", + "--execute", + ) + + added = _add_todo( + registry, + override_runtime, + goal_id=goal_id, + text="Every hook of this call shares one runtime root.", + ) + todo_id = str(added["todo_id"]) + lease = _cli( + registry, + override_runtime, + "task-lease", + "acquire", + "--goal-id", + goal_id, + "--todo-id", + todo_id, + "--owner", + "agent-a", + "--idempotency-key", + "one-root-lease", + "--ttl-seconds", + "120", + ) + assert lease["acquired"] is True + updated = _cli( + registry, + override_runtime, + "todo", + "update", + "--goal-id", + goal_id, + "--todo-id", + todo_id, + "--note", + "Observed under the override root.", + "--agent-id", + "agent-a", + ) + assert updated["changed"] is True + followups = _cli( + registry, + override_runtime, + "todo", + "capture-followups", + "--goal-id", + goal_id, + "--follow-up", + "Verify that one goal keeps one candidate lineage.", + "--evidence", + "validation://one-root-followup", + ) + assert followups["changed"] is True + completed = _cli( + registry, + override_runtime, + "todo", + "complete", + "--goal-id", + goal_id, + "--todo-id", + todo_id, + "--agent-id", + "agent-a", + "--task-lease-idempotency-key", + "one-root-lease", + "--task-lease-expected-version", + str(lease["lease"]["version"]), # type: ignore[index] + "--evidence", + "validation://one-root-complete", + "--no-follow-up", + ) + assert completed["completed"] is True + + responses = { + "todo add": added, + "task-lease acquire": lease, + "todo update": updated, + "todo capture-followups": followups, + "todo complete": completed, + } + for label, payload in responses.items(): + evidence = payload["authority_shadow"] + assert evidence["outcome"] == "captured", label # type: ignore[index] + identities = {payload["authority_shadow"]["store_identity"] for payload in responses.values()} # type: ignore[index] + assert len(identities) == 1 + store_path, store = _store_document(override_runtime, goal_id) + assert store["store_identity"] == identities.pop() + assert store["cursor"] == str(len(responses)) + head_todo_ids = {todo["todo_id"] for todo in store["head"]["todos"]} # type: ignore[index] + assert todo_id in head_todo_ids + assert len(head_todo_ids) == 2 + [lease_record] = store["head"]["leases"] # type: ignore[index] + assert lease_record["todo_id"] == todo_id + assert lease_record["status"] == "released" + assert (override_runtime / "goals" / goal_id / "task-leases" / f"{todo_id}.json").exists() + assert not (registry_runtime / "authority-shadow").exists() + assert not (registry_runtime / "goals" / goal_id / "task-leases").exists() + assert store_path.is_relative_to(override_runtime) + assert todo_id in state.read_text(encoding="utf-8") diff --git a/tests/control_plane/test_local_authority_shadow_config.py b/tests/control_plane/test_local_authority_shadow_config.py new file mode 100644 index 0000000000..dc8c8a6b32 --- /dev/null +++ b/tests/control_plane/test_local_authority_shadow_config.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from loopx.cli import main +from loopx.configure_goal import configure_goal + + +GOAL_ID = "local-authority-shadow-config" +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def _registry(tmp_path: Path) -> Path: + state = tmp_path / "ACTIVE_GOAL_STATE.md" + state.write_text( + "---\n" + f"goal_id: {GOAL_ID}\n" + "handoff_mode: hard_lease\n" + "---\n\n" + "## Agent Todo\n\n", + encoding="utf-8", + ) + registry = tmp_path / "registry.json" + registry.write_text( + json.dumps( + { + "goals": [ + { + "id": GOAL_ID, + "repo": str(tmp_path), + "state_file": state.name, + "coordination": { + "agent_model": "peer_v1", + "registered_agents": ["agent-a", "agent-b"], + }, + } + ] + } + ), + encoding="utf-8", + ) + return registry + + +def test_configure_goal_enables_and_clears_closed_file_shadow_config( + tmp_path: Path, +) -> None: + registry = _registry(tmp_path) + + preview = configure_goal( + registry_path=registry, + goal_id=GOAL_ID, + local_authority_shadow_file=True, + execute=False, + ) + + assert preview["changed_fields"] == ["local_authority_shadow"] + assert preview["before"]["local_authority_shadow"] == { + "enabled": False, + "mode": None, + "status": "disabled", + } + assert preview["after"]["local_authority_shadow"] == { + "enabled": True, + "mode": "file_one_way", + "status": "enabled", + } + + applied = configure_goal( + registry_path=registry, + goal_id=GOAL_ID, + local_authority_shadow_file=True, + execute=True, + ) + assert applied["written"] is True + goal = json.loads(registry.read_text(encoding="utf-8"))["goals"][0] + assert goal["coordination"]["authority_shadow"] == { + "schema_version": "loopx_local_authority_shadow_config_v0", + "mode": "file_one_way", + } + assert goal["coordination"]["agent_model"] == "peer_v1" + + repeated = configure_goal( + registry_path=registry, + goal_id=GOAL_ID, + local_authority_shadow_file=True, + execute=True, + ) + assert repeated["written"] is False + + cleared = configure_goal( + registry_path=registry, + goal_id=GOAL_ID, + clear_local_authority_shadow=True, + execute=True, + ) + assert cleared["changed_fields"] == ["local_authority_shadow"] + goal = json.loads(registry.read_text(encoding="utf-8"))["goals"][0] + assert "authority_shadow" not in goal["coordination"] + assert goal["coordination"]["registered_agents"] == ["agent-a", "agent-b"] + + +def test_configure_goal_rejects_enable_and_clear_in_one_operation( + tmp_path: Path, +) -> None: + registry = _registry(tmp_path) + with pytest.raises(ValueError, match="cannot be combined"): + configure_goal( + registry_path=registry, + goal_id=GOAL_ID, + local_authority_shadow_file=True, + clear_local_authority_shadow=True, + ) + + +def test_configure_goal_cli_exposes_default_off_shadow_boundary( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + registry = _registry(tmp_path) + + exit_code = main( + [ + "--registry", + str(registry), + "--runtime-root", + str(tmp_path / "runtime"), + "--format", + "json", + "configure-goal", + "--goal-id", + GOAL_ID, + "--local-authority-shadow-file", + ] + ) + + assert exit_code == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["dry_run"] is True + assert payload["written"] is False + assert payload["after"]["local_authority_shadow"]["enabled"] is True + feature = next( + item + for item in payload["configuration_catalog"]["features"] + if item["feature_id"] == "local_authority_shadow" + ) + assert feature["display_name"] == "Local post-commit authority observation" + assert feature["availability"] == "experimental_opt_in" + assert "parity" not in feature["consider_when"].lower() + assert "post-commit snapshot" in feature["effect"] + assert feature["does_not"] == [ + "read the candidate for lifecycle decisions", + "write candidate state back into Markdown or task-lease files", + "promote shared authority or fence legacy writers", + "bind the snapshot to the exact primary transaction", + "guarantee delivery through a durable outbox", + "compare source and candidate or issue a parity verdict", + ] + assert feature["commands"]["apply_disable"].endswith( + "--clear-local-authority-shadow --execute" + ) + assert "authority_shadow" not in json.loads( + registry.read_text(encoding="utf-8") + )["goals"][0]["coordination"] + + +def test_rfc_disambiguates_historical_and_current_stage_numbering() -> None: + english = ( + REPO_ROOT + / "docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md" + ).read_text(encoding="utf-8") + chinese = ( + REPO_ROOT + / "docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md" + ).read_text(encoding="utf-8") + + assert "historical #3669 implementation sequence" in english + assert "part of the Stage 0 reference foundation" in english + assert "not the Stage 3 remote-shadow phase in Section 11" in english + assert "#3669 历史实施序列" in chinese + assert "属于 Stage 0 reference foundation" in chinese + assert "不是第 11 节的 Stage 3 远端 shadow 阶段" in chinese diff --git a/tests/control_plane/test_local_authority_shadow_runtime.py b/tests/control_plane/test_local_authority_shadow_runtime.py new file mode 100644 index 0000000000..614045a959 --- /dev/null +++ b/tests/control_plane/test_local_authority_shadow_runtime.py @@ -0,0 +1,567 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from loopx.control_plane.coordination.local_authority_shadow_adapter import ( + LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA, + observe_local_authority_commit, +) +from loopx.control_plane.todos.handoff_mode import set_goal_handoff_mode +from loopx.control_plane.work_items.task_lease import ( + acquire_task_lease, + release_task_lease, + renew_task_lease, + transfer_task_lease, +) +from loopx.event_sourced_state import ( + TODO_ADDED, + AppendOnlyStateEventStore, + make_state_event, +) +from loopx.todo_followups import capture_followup_todos +from loopx.todos import ( + add_goal_todo, + archive_completed_todos, + complete_goal_todo, + supersede_goal_todo, + update_goal_todo, +) + + +GOAL_ID = "goal-shadow" +AGENT_A = "agent-a" +AGENT_B = "agent-b" + + +def _fixture(tmp_path: Path, *, enabled: bool) -> tuple[Path, Path, Path]: + repo = tmp_path / "repo" + repo.mkdir() + state = repo / "ACTIVE_GOAL_STATE.md" + state.write_text( + "---\n" + f"goal_id: {GOAL_ID}\n" + "handoff_mode: hard_lease\n" + "updated_at: 2026-09-02T00:00:00+00:00\n" + "---\n\n" + "## Agent Todo\n\n", + encoding="utf-8", + ) + runtime_root = tmp_path / "runtime" + coordination: dict[str, object] = { + "agent_model": "peer_v1", + "registered_agents": [AGENT_A, AGENT_B], + } + if enabled: + coordination["authority_shadow"] = { + "schema_version": "loopx_local_authority_shadow_config_v0", + "mode": "file_one_way", + } + registry = tmp_path / "registry.json" + registry.write_text( + json.dumps( + { + "common_runtime_root": str(runtime_root), + "goals": [ + { + "id": GOAL_ID, + "domain": "harness_self_improvement", + "status": "active", + "repo": str(repo), + "state_file": state.name, + "adapter": {"kind": "harness_self_improvement"}, + "coordination": coordination, + } + ], + } + ), + encoding="utf-8", + ) + return registry, state, runtime_root + + +def _add(registry: Path) -> dict: + return add_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + role="agent", + text="Exercise one-way local authority shadowing.", + task_class="advancement_task", + ) + + +def _shadow_document(runtime_root: Path) -> dict: + paths = list( + ( + runtime_root + / "authority-shadow" + / "file" + / GOAL_ID + ).glob("authority-store-*.json") + ) + assert len(paths) == 1 + return json.loads(paths[0].read_text(encoding="utf-8")) + + +def test_default_off_public_writers_never_call_shadow_runtime( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry, _state, runtime_root = _fixture(tmp_path, enabled=False) + calls: list[object] = [] + + def forbidden(*args: object, **kwargs: object) -> object: + calls.append((args, kwargs)) + raise AssertionError("default-off path constructed the shadow runtime") + + monkeypatch.setattr( + "loopx.control_plane.coordination.local_authority_shadow_adapter.effect_runtime_result", + forbidden, + ) + + result = _add(registry) + lease_result = acquire_task_lease( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + todo_id=str(result["todo_id"]), + owner=AGENT_A, + idempotency_key="default-off", + ttl_seconds=120, + ) + + assert result["ok"] is True + assert lease_result["ok"] is True + assert "authority_shadow" not in result + assert "authority_shadow" not in lease_result + assert calls == [] + assert not (runtime_root / "authority-shadow").exists() + + +def test_enabled_todo_public_facades_emit_post_commit_evidence(tmp_path: Path) -> None: + registry, _state, runtime_root = _fixture(tmp_path, enabled=True) + + added = _add(registry) + todo_id = str(added["todo_id"]) + acquired = acquire_task_lease( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + todo_id=todo_id, + owner=AGENT_A, + idempotency_key="todo-terminal-a", + ttl_seconds=120, + ) + updated = update_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=todo_id, + note="A public-safe update.", + agent_id=AGENT_A, + ) + completed = complete_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=todo_id, + role="agent", + no_followup=True, + agent_id=AGENT_A, + task_lease_idempotency_key="todo-terminal-a", + task_lease_expected_version=int(acquired["lease"]["version"]), + ) + archived = archive_completed_todos( + registry_path=registry, + goal_id=GOAL_ID, + role="agent", + max_active_done=0, + dry_run=False, + ) + replacement = _add(registry) + replacement_lease = acquire_task_lease( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + todo_id=str(replacement["todo_id"]), + owner=AGENT_A, + idempotency_key="todo-terminal-b", + ttl_seconds=120, + ) + superseded = supersede_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=str(replacement["todo_id"]), + role="agent", + reason="Replace obsolete work.", + next_agent_todo="Carry the bounded work forward.", + agent_id=AGENT_A, + task_lease_idempotency_key="todo-terminal-b", + task_lease_expected_version=int(replacement_lease["lease"]["version"]), + ) + + for result in (added, updated, completed, archived, replacement, superseded): + assert result["ok"] is True + assert result["authority_shadow"]["schema_version"] == ( + LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA + ) + assert result["authority_shadow"]["primary_writeback_preserved"] is True + assert result["authority_shadow"]["provider_to_local_writes"] is False + assert result["authority_shadow"]["capture_kind"] == "post_commit_snapshot" + assert result["authority_shadow"]["source_transaction_correlated"] is False + assert result["authority_shadow"]["durable_source_outbox"] is False + assert result["authority_shadow"]["source_candidate_compared"] is False + assert result["authority_shadow"]["parity_verdict"] == "not_evaluated" + + +def test_enabled_task_lease_facades_shadow_only_committed_mutations( + tmp_path: Path, +) -> None: + registry, _state, runtime_root = _fixture(tmp_path, enabled=True) + todo_id = str(_add(registry)["todo_id"]) + + acquired = acquire_task_lease( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + todo_id=todo_id, + owner=AGENT_A, + idempotency_key="lease-a", + ttl_seconds=120, + ) + replayed_acquire = acquire_task_lease( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + todo_id=todo_id, + owner=AGENT_A, + idempotency_key="lease-a", + ttl_seconds=120, + ) + renewed = renew_task_lease( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + todo_id=todo_id, + owner=AGENT_A, + idempotency_key="lease-a", + expected_version=1, + ttl_seconds=120, + ) + transferred = transfer_task_lease( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + todo_id=todo_id, + owner=AGENT_A, + idempotency_key="lease-a", + new_owner=AGENT_B, + new_idempotency_key="lease-b", + expected_version=2, + ttl_seconds=120, + ) + released = release_task_lease( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + todo_id=todo_id, + owner=AGENT_B, + idempotency_key="lease-b", + expected_version=3, + ) + + for result in (acquired, renewed, transferred, released): + assert result["authority_shadow"]["outcome"] in { + "captured", + "replayed", + "ambiguous_reconciled", + } + assert replayed_acquire["idempotent"] is True + assert "authority_shadow" not in replayed_acquire + + +def test_handoff_mode_and_direct_followup_writers_refresh_the_same_shadow( + tmp_path: Path, +) -> None: + registry, _state, runtime_root = _fixture(tmp_path, enabled=True) + + mode = set_goal_handoff_mode( + registry_path=registry, + goal_id=GOAL_ID, + mode="legacy", + ) + followups = capture_followup_todos( + registry_path=registry, + goal_id=GOAL_ID, + followups=["Verify the migrated authority projection."], + evidence="validation://local-shadow-followup", + ) + + assert mode["changed"] is True + assert mode["authority_shadow"]["outcome"] == "captured" + assert followups["changed"] is True + assert followups["authority_shadow"]["outcome"] == "captured" + head = _shadow_document(runtime_root)["head"] + assert head["handoff_mode"] == "legacy" + assert [todo["todo_id"] for todo in head["todos"]] == [ + followups["items"][0]["todo_id"] + ] + + +def test_event_projected_completion_refreshes_shadow_after_releasing_lease( + tmp_path: Path, +) -> None: + registry, state, runtime_root = _fixture(tmp_path, enabled=True) + todo_id = "todo_event_shadow" + AppendOnlyStateEventStore(state.with_name("events.jsonl")).append( + make_state_event( + event_id="evt-event-shadow-parent", + goal_id=GOAL_ID, + event_type=TODO_ADDED, + refs={"todo_id": todo_id}, + payload={ + "role": "agent", + "title": "Complete the event-projected shadow task.", + "task_class": "advancement_task", + "claimed_by": AGENT_A, + }, + recorded_at="2026-09-02T00:00:00+00:00", + ) + ) + lease_key = "event-shadow-instance" + acquire_task_lease( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + todo_id=todo_id, + owner=AGENT_A, + idempotency_key=lease_key, + ttl_seconds=120, + ) + + completed = complete_goal_todo( + registry_path=registry, + goal_id=GOAL_ID, + todo_id=todo_id, + claimed_by=AGENT_A, + agent_id=AGENT_A, + task_lease_idempotency_key=lease_key, + task_lease_expected_version=1, + evidence="validation://event-shadow-completion", + no_followup=True, + ) + + assert completed["source"] == "event_log" + assert completed["changed"] is True + assert completed["authority_shadow"]["outcome"] == "captured" + head = _shadow_document(runtime_root)["head"] + assert len(head["leases"]) == 1 + assert head["leases"][0]["todo_id"] == todo_id + assert head["leases"][0]["status"] == "released" + projected = next(todo for todo in head["todos"] if todo["todo_id"] == todo_id) + assert projected["status"] == "done" + + +def test_candidate_failure_never_changes_committed_todo_result( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry, state, _runtime_root = _fixture(tmp_path, enabled=True) + + def unavailable( + _method: str, + params: dict[str, object], + **_kwargs: object, + ) -> dict[str, object]: + return { + "schema_version": LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA, + "outcome": "unavailable", + "reason_code": "injected_outage", + "goal_id": params["goal_id"], + "observation_id": params["observation_id"], + "source_digest": params["source_digest"], + "primary_authority": "legacy_local", + "candidate_provider": "file", + "candidate_read_for_decision": False, + "provider_to_local_writes": False, + "primary_writeback_preserved": True, + "capture_kind": "post_commit_snapshot", + "source_transaction_correlated": False, + "durable_source_outbox": False, + "source_candidate_compared": False, + "parity_verdict": "not_evaluated", + "store_identity": None, + "provider_revision": None, + "cursor": None, + } + + monkeypatch.setattr( + "loopx.control_plane.coordination.local_authority_shadow_adapter.effect_runtime_result", + unavailable, + ) + + result = _add(registry) + + assert result["ok"] is True + assert result["added"] is True + assert result["authority_shadow"]["outcome"] == "unavailable" + assert str(result["todo_id"]) in state.read_text(encoding="utf-8") + + +def test_invalid_shadow_config_is_typed_but_preserves_primary_write( + tmp_path: Path, +) -> None: + registry, state, _runtime_root = _fixture(tmp_path, enabled=True) + payload = json.loads(registry.read_text(encoding="utf-8")) + payload["goals"][0]["coordination"]["authority_shadow"]["mode"] = "remote" + registry.write_text(json.dumps(payload), encoding="utf-8") + + result = _add(registry) + + assert result["ok"] is True + assert result["authority_shadow"]["outcome"] == "failed" + assert result["authority_shadow"]["reason_code"] == "invalid_shadow_config" + assert str(result["todo_id"]) in state.read_text(encoding="utf-8") + + +def test_observer_is_default_off_without_creating_lock_or_provider_directory( + tmp_path: Path, +) -> None: + registry, _state, runtime_root = _fixture(tmp_path, enabled=False) + + assert ( + observe_local_authority_commit( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + observation_trigger="todo_update", + ) + is None + ) + assert not (runtime_root / "authority-shadow").exists() + + +def test_provider_revision_conflict_resamples_source_under_same_observation_lock( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry, _state, runtime_root = _fixture(tmp_path, enabled=True) + projections = iter( + ( + { + "schema_version": "loopx_local_authority_shadow_projection_v0", + "goal_id": GOAL_ID, + "handoff_mode": "hard_lease", + "todos": [{"todo_id": "todo_old", "status": "open"}], + "leases": [], + }, + { + "schema_version": "loopx_local_authority_shadow_projection_v0", + "goal_id": GOAL_ID, + "handoff_mode": "hard_lease", + "todos": [{"todo_id": "todo_fresh", "status": "open"}], + "leases": [], + }, + ) + ) + requests: list[dict[str, object]] = [] + + monkeypatch.setattr( + "loopx.control_plane.coordination.local_authority_shadow_adapter._stable_projection", + lambda **_kwargs: next(projections), + ) + + def conflict_then_advance( + _method: str, + params: dict[str, object], + **_kwargs: object, + ) -> dict[str, object]: + requests.append(dict(params)) + outcome = "conflict_retry_required" if len(requests) == 1 else "captured" + return { + "schema_version": LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA, + "outcome": outcome, + "reason_code": ( + "provider_revision_mismatch" if len(requests) == 1 else None + ), + "goal_id": GOAL_ID, + "observation_id": params["observation_id"], + "source_digest": params["source_digest"], + "primary_authority": "legacy_local", + "candidate_provider": "file", + "candidate_read_for_decision": False, + "provider_to_local_writes": False, + "primary_writeback_preserved": True, + "capture_kind": "post_commit_snapshot", + "source_transaction_correlated": False, + "durable_source_outbox": False, + "source_candidate_compared": False, + "parity_verdict": "not_evaluated", + "store_identity": "file:test", + "provider_revision": "file:2:test", + "cursor": "2", + } + + monkeypatch.setattr( + "loopx.control_plane.coordination.local_authority_shadow_adapter.effect_runtime_result", + conflict_then_advance, + ) + + result = observe_local_authority_commit( + registry_path=registry, + runtime_root=runtime_root, + goal_id=GOAL_ID, + observation_trigger="todo_update:todo_a:now", + ) + + assert result is not None + assert result["outcome"] == "captured" + assert len(requests) == 2 + assert requests[0]["source_digest"] != requests[1]["source_digest"] + assert requests[0]["observation_id"] != requests[1]["observation_id"] + assert all(request["runtime_root"] == str(runtime_root) for request in requests) + assert all("provider_directory" not in request for request in requests) + + +def test_relative_common_runtime_root_resolves_against_the_project_root_for_every_hook( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from loopx.control_plane.work_items.task_lease import runtime_root_from_registry + from loopx.paths import registry_project_root + + registry, _state, _absolute_runtime = _fixture(tmp_path, enabled=True) + document = json.loads(registry.read_text(encoding="utf-8")) + document["common_runtime_root"] = "runtime-relative" + registry.write_text(json.dumps(document), encoding="utf-8") + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir() + monkeypatch.chdir(elsewhere) + expected_root = registry_project_root(registry) / "runtime-relative" + + added = _add(registry) + lease = acquire_task_lease( + registry_path=registry, + runtime_root=runtime_root_from_registry(registry, None), + goal_id=GOAL_ID, + todo_id=str(added["todo_id"]), + owner=AGENT_A, + idempotency_key="relative-root", + ttl_seconds=120, + ) + handoff = set_goal_handoff_mode( + registry_path=registry, + goal_id=GOAL_ID, + mode="hard_lease", + ) + + assert added["authority_shadow"]["outcome"] == "captured" + assert lease["authority_shadow"]["outcome"] == "captured" + assert handoff["changed"] is False + assert added["authority_shadow"]["store_identity"] == lease["authority_shadow"]["store_identity"] + assert (expected_root / "authority-shadow" / "file" / GOAL_ID).is_dir() + assert (expected_root / "goals" / GOAL_ID / "task-leases" / f"{added['todo_id']}.json").exists() + assert not (elsewhere / "runtime-relative").exists() + document = _shadow_document(expected_root) + assert document["cursor"] == "2" + assert len(document["head"]["leases"]) == 1 diff --git a/tests/control_plane/test_state_migration_authority_shadow.py b/tests/control_plane/test_state_migration_authority_shadow.py new file mode 100644 index 0000000000..6f8f1efaf9 --- /dev/null +++ b/tests/control_plane/test_state_migration_authority_shadow.py @@ -0,0 +1,262 @@ +from __future__ import annotations + +import json +from collections.abc import Iterator +from pathlib import Path + +import pytest + +from loopx.state_migration import migrate_legacy_state, render_state_migration_markdown + + +OLD_GOAL_ID = "legacy-goal" +NEW_GOAL_ID = "migrated-goal" +OLD_STORE_IDENTITY = "file:11111111111111111111111111111111" + + +def _migration_fixture(tmp_path: Path) -> dict[str, Path]: + legacy_runtime = tmp_path / "legacy-runtime" + target_runtime = tmp_path / "target-runtime" + source_repo = tmp_path / "legacy-repo" + target_repo = tmp_path / "target-repo" + source_repo.mkdir() + target_repo.mkdir() + + source_state = source_repo / "ACTIVE_GOAL_STATE.md" + source_state.write_text( + "---\n" + f"goal_id: {OLD_GOAL_ID}\n" + "handoff_mode: soft_claim\n" + "updated_at: 2026-09-02T00:00:00+10:00\n" + "---\n\n" + "## Agent Todo\n\n" + "- [ ] Preserve the new local authority only.\n", + encoding="utf-8", + ) + + legacy_registry = tmp_path / "legacy-registry.json" + legacy_registry.write_text( + json.dumps( + { + "schema_version": "0.1", + "common_runtime_root": str(legacy_runtime), + "goals": [ + { + "id": OLD_GOAL_ID, + "status": "active", + "repo": str(source_repo), + "state_file": source_state.name, + "coordination": { + "agent_model": "peer_v1", + "registered_agents": ["agent-a", "agent-b"], + "authority_shadow": { + "schema_version": ( + "loopx_local_authority_shadow_config_v0" + ), + "mode": "file_one_way", + }, + }, + } + ], + } + ), + encoding="utf-8", + ) + + source_goal_runtime = legacy_runtime / "goals" / OLD_GOAL_ID + (source_goal_runtime / "task-leases").mkdir(parents=True) + (source_goal_runtime / "task-leases" / "safe-local.json").write_text( + json.dumps( + { + "goal_id": OLD_GOAL_ID, + "todo_id": "safe-local", + "owner": "agent-a", + "version": 1, + "lease_epoch": 1, + "status": "released", + } + ), + encoding="utf-8", + ) + source_shadow_store = ( + legacy_runtime / "authority-shadow" / "file" / OLD_GOAL_ID + ) + source_shadow_store.mkdir(parents=True) + (source_shadow_store / "store-identity").write_text( + OLD_STORE_IDENTITY, + encoding="utf-8", + ) + (source_shadow_store / "authority-store-legacy.json").write_text( + json.dumps( + { + "goal_id": OLD_GOAL_ID, + "store_identity": OLD_STORE_IDENTITY, + "provider_revision": "file:99:legacy-lineage", + "cursor": "99", + "private_provider_byte": "must-never-migrate", + "source_path": str(source_repo), + } + ), + encoding="utf-8", + ) + + return { + "legacy_registry": legacy_registry, + "target_registry": tmp_path / "target-registry.json", + "legacy_runtime": legacy_runtime, + "target_runtime": target_runtime, + "source_repo": source_repo, + "target_repo": target_repo, + } + + +def _migrate(paths: dict[str, Path], *, execute: bool) -> dict[str, object]: + return migrate_legacy_state( + legacy_registry_path=paths["legacy_registry"], + target_registry_path=paths["target_registry"], + legacy_runtime_root=paths["legacy_runtime"], + target_runtime_root=paths["target_runtime"], + goal_ids=[OLD_GOAL_ID], + goal_id_map={OLD_GOAL_ID: NEW_GOAL_ID}, + path_map={str(paths["source_repo"]): str(paths["target_repo"])}, + copy_active_state=True, + copy_runtime=True, + execute=execute, + ) + + +def test_execute_excludes_old_shadow_and_seeds_fresh_target_lineage( + tmp_path: Path, +) -> None: + paths = _migration_fixture(tmp_path) + + result = _migrate(paths, execute=True) + + assert result["ok"] is True + runtime_result = result["runtime_goals"][0] # type: ignore[index] + assert runtime_result["copied"] is True + + target_goal_runtime = paths["target_runtime"] / "goals" / NEW_GOAL_ID + copied_lease = json.loads( + (target_goal_runtime / "task-leases" / "safe-local.json").read_text( + encoding="utf-8" + ) + ) + assert copied_lease["goal_id"] == NEW_GOAL_ID + + target_store_dir = ( + paths["target_runtime"] / "authority-shadow" / "file" / NEW_GOAL_ID + ) + target_identity = (target_store_dir / "store-identity").read_text(encoding="utf-8") + assert target_identity.startswith("file:") + assert target_identity != OLD_STORE_IDENTITY + + store_paths = list(target_store_dir.glob("authority-store-*.json")) + assert len(store_paths) == 1 + store_payload = json.loads(store_paths[0].read_text(encoding="utf-8")) + assert store_payload["goal_id"] == NEW_GOAL_ID + assert store_payload["store_identity"] == target_identity + assert store_payload["cursor"] == "1" + assert len(store_payload["committed"]) == 1 + serialized_store = json.dumps(store_payload, sort_keys=True) + assert OLD_STORE_IDENTITY not in serialized_store + assert "file:99:legacy-lineage" not in serialized_store + assert str(paths["source_repo"]) not in serialized_store + assert "must-never-migrate" not in serialized_store + + seed = result["authority_shadow_seeds"][0] # type: ignore[index] + assert seed["goal_id"] == NEW_GOAL_ID + assert seed["attempted"] is True + assert seed["outcome"] == "captured" + + +def test_dry_run_reports_exclusion_and_seed_plan_without_writing( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + paths = _migration_fixture(tmp_path) + sentinel = b'{"schema_version":"existing","goals":[]}\n' + paths["target_registry"].write_bytes(sentinel) + + def forbidden_observer(**_kwargs: object) -> object: + raise AssertionError("dry-run called the shadow observer") + + original_rglob = Path.rglob + + def guarded_rglob(path: Path, pattern: str) -> Iterator[Path]: + if path.name == "authority-shadow": + raise AssertionError("migration entered candidate-provider storage") + return original_rglob(path, pattern) + + monkeypatch.setattr( + "loopx.control_plane.coordination.local_authority_shadow_adapter." + "observe_local_authority_commit", + forbidden_observer, + ) + monkeypatch.setattr(Path, "rglob", guarded_rglob) + + result = _migrate(paths, execute=False) + + assert result["ok"] is True + assert result["dry_run"] is True + assert paths["target_registry"].read_bytes() == sentinel + assert not paths["target_runtime"].exists() + runtime_result = result["runtime_goals"][0] # type: ignore[index] + assert runtime_result["copied_file_count"] == 0 + seed = result["authority_shadow_seeds"][0] # type: ignore[index] + assert seed == { + "schema_version": "loopx_state_migration_shadow_seed_evidence_v0", + "goal_id": NEW_GOAL_ID, + "attempted": False, + "outcome": "planned", + "reason_code": None, + } + rendered = render_state_migration_markdown(result) + assert "Authority Shadow Seeds" in rendered + assert "outcome=`planned`" in rendered + assert "must-never-migrate" not in json.dumps(result, sort_keys=True) + assert "must-never-migrate" not in rendered + + +def test_seed_failure_is_public_safe_evidence_and_does_not_reverse_migration( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + paths = _migration_fixture(tmp_path) + + def fail_observer(**_kwargs: object) -> object: + raise RuntimeError("credential=private-provider-value") + + monkeypatch.setattr( + "loopx.control_plane.coordination.local_authority_shadow_adapter." + "observe_local_authority_commit", + fail_observer, + ) + + result = _migrate(paths, execute=True) + + assert result["ok"] is True + assert result["wrote_project_registry"] is True + migrated_registry = json.loads(paths["target_registry"].read_text(encoding="utf-8")) + assert migrated_registry["goals"][0]["id"] == NEW_GOAL_ID + assert ( + paths["target_runtime"] + / "goals" + / NEW_GOAL_ID + / "task-leases" + / "safe-local.json" + ).exists() + + seed = result["authority_shadow_seeds"][0] # type: ignore[index] + assert seed["outcome"] == "failed" + assert seed["reason_code"] == "post_migration_shadow_seed_failed" + assert seed["attempted"] is True + assert "credential" not in json.dumps(result, sort_keys=True) + assert "private-provider-value" not in json.dumps(result, sort_keys=True) + assert not ( + paths["target_runtime"] + / "authority-shadow" + / "file" + / NEW_GOAL_ID + / "authority-store-legacy.json" + ).exists() diff --git a/tests/control_plane_ts/local_authority_shadow.test.ts b/tests/control_plane_ts/local_authority_shadow.test.ts new file mode 100644 index 0000000000..a814514388 --- /dev/null +++ b/tests/control_plane_ts/local_authority_shadow.test.ts @@ -0,0 +1,242 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +import type { + AuthorityStore, + AuthorityStoreCommit, + AuthorityStoreCommitResult, + AuthorityStoreIdentityResult, + AuthorityStoreLoadResult, + AuthorityStoreReceiptResult, + AuthorityStoreScanResult, +} from "../../loopx/control_plane/coordination/authority_store.ts"; +import { FileAuthorityStore } from "../../loopx/control_plane/coordination/file_authority_store.ts"; +import { + LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA, + recordLocalAuthorityShadow, +} from "../../loopx/control_plane/coordination/local_authority_shadow.ts"; + +function request(directory: string, operationId = "local-operation-a") { + return { + schema_version: "loopx_local_authority_shadow_request_v0", + mode: "file_one_way", + runtime_root: directory, + goal_id: "goal-a", + observation_id: operationId, + observation_trigger: "todo_update", + source_digest: `sha256:${"a".repeat(64)}`, + source_projection: { + schema_version: "loopx_local_authority_shadow_projection_v0", + goal_id: "goal-a", + handoff_mode: "hard_lease", + todos: [{ todo_id: "todo-a", status: "open", claimed_by: "agent-a" }], + leases: [{ todo_id: "todo-a", version: 2, lease_epoch: 1, status: "active" }], + }, + }; +} + +test("one-way file shadow captures a post-commit observation without claiming parity", async (t) => { + const root = await mkdtemp(join(tmpdir(), "loopx-local-authority-shadow-")); + t.after(() => rm(root, { recursive: true, force: true })); + + const evidence = await recordLocalAuthorityShadow(request(root)); + + assert.equal(evidence.schema_version, LOCAL_AUTHORITY_SHADOW_EVIDENCE_SCHEMA); + assert.equal(evidence.outcome, "captured"); + assert.equal(evidence.primary_authority, "legacy_local"); + assert.equal(evidence.candidate_read_for_decision, false); + assert.equal(evidence.provider_to_local_writes, false); + assert.equal(evidence.primary_writeback_preserved, true); + assert.equal(evidence.capture_kind, "post_commit_snapshot"); + assert.equal(evidence.source_transaction_correlated, false); + assert.equal(evidence.durable_source_outbox, false); + assert.equal(evidence.source_candidate_compared, false); + assert.equal(evidence.parity_verdict, "not_evaluated"); + const loaded = await new FileAuthorityStore( + join(root, "authority-shadow", "file", "goal-a"), + "goal-a", + ).loadAuthority(); + assert.equal(loaded.status, "loaded"); + if (loaded.status === "loaded") { + assert.deepEqual(loaded.head, request(root).source_projection); + } +}); + +test("same observation reuses its candidate-side capture receipt", async (t) => { + const root = await mkdtemp(join(tmpdir(), "loopx-local-authority-shadow-")); + t.after(() => rm(root, { recursive: true, force: true })); + + assert.equal((await recordLocalAuthorityShadow(request(root))).outcome, "captured"); + const replay = await recordLocalAuthorityShadow(request(root)); + + assert.equal(replay.outcome, "replayed"); + const page = await new FileAuthorityStore( + join(root, "authority-shadow", "file", "goal-a"), + "goal-a", + ).scanCommitted(null, 10); + assert.equal(page.status, "page"); + if (page.status === "page") assert.equal(page.transactions.length, 1); +}); + +test("observation trigger does not imply exact source-transaction binding", async (t) => { + const root = await mkdtemp(join(tmpdir(), "loopx-local-authority-shadow-")); + t.after(() => rm(root, { recursive: true, force: true })); + const requestAfterConcurrentCommit = request(root); + requestAfterConcurrentCommit.observation_trigger = "todo_add:todo-a"; + requestAfterConcurrentCommit.source_projection.todos.push({ + todo_id: "todo-b", + status: "open", + claimed_by: "agent-b", + }); + + const observation = await recordLocalAuthorityShadow(requestAfterConcurrentCommit); + + assert.equal(observation.outcome, "captured"); + assert.equal(observation.source_transaction_correlated, false); + assert.equal(observation.parity_verdict, "not_evaluated"); + const page = await new FileAuthorityStore( + join(root, "authority-shadow", "file", "goal-a"), + "goal-a", + ).scanCommitted(null, 10); + assert.equal(page.status, "page"); + if (page.status === "page") { + assert.equal(page.transactions[0]?.events[0]?.observation_trigger, "todo_add:todo-a"); + const capturedTodos = page.transactions[0]?.projection.todos; + assert.ok(Array.isArray(capturedTodos)); + assert.equal(capturedTodos.length, 2); + } +}); + +test("legacy source_operation wording is rejected by the closed observation contract", async () => { + const legacyRequest = { ...request("/not-used") } as Record; + legacyRequest.source_operation = legacyRequest.observation_trigger; + delete legacyRequest.observation_trigger; + + await assert.rejects( + recordLocalAuthorityShadow(legacyRequest), + /unsupported fields: source_operation/u, + ); +}); + +class UnavailableStore implements AuthorityStore { + commits = 0; + + async storeIdentity(): Promise { + return { status: "unavailable", reason_code: "injected", reason: "offline" }; + } + async loadAuthority(): Promise { + return { status: "unavailable", reason_code: "injected", reason: "offline" }; + } + async commitAuthority(_commit: AuthorityStoreCommit): Promise { + this.commits += 1; + return { status: "failed", reason_code: "unexpected", reason: "must not commit" }; + } + async readReceipt(_operationId: string): Promise { + return { status: "missing" }; + } + async scanCommitted(_afterCursor: string | null, _limit: number): Promise { + return { status: "page", transactions: [], next_cursor: null, has_more: false }; + } +} + +test("candidate unavailability is typed evidence and attempts no commit", async () => { + const store = new UnavailableStore(); + const evidence = await recordLocalAuthorityShadow(request("/not-used"), { + openStore: () => store, + }); + + assert.equal(evidence.outcome, "unavailable"); + assert.equal(evidence.reason_code, "injected"); + assert.equal(evidence.primary_writeback_preserved, true); + assert.equal(store.commits, 0); +}); + +class RevisionConflictStore extends UnavailableStore { + override async storeIdentity(): Promise { + return { status: "available", store_identity: "file:test" }; + } + override async loadAuthority(): Promise { + return { + status: "loaded", + head: {}, + provider_revision: "provider-revision-a", + cursor: "1", + }; + } + override async commitAuthority( + _commit: AuthorityStoreCommit, + ): Promise { + this.commits += 1; + return { + status: "conflict", + conflict_kind: "provider_revision_mismatch", + current_provider_revision: "provider-revision-b", + current_cursor: "2", + }; + } +} + +test("provider revision conflict requests a fresh source projection without stale retry", async () => { + const store = new RevisionConflictStore(); + + const result = await recordLocalAuthorityShadow(request("/not-used"), { + openStore: () => store, + }); + + assert.equal(result.outcome, "conflict_retry_required"); + assert.equal(result.reason_code, "provider_revision_mismatch"); + assert.equal(store.commits, 1); +}); + +test("goal id cannot escape the fixed shadow directory", async () => { + await assert.rejects( + recordLocalAuthorityShadow({ + ...request("/not-used"), + goal_id: "../other-goal", + source_projection: { + ...request("/not-used").source_projection, + goal_id: "../other-goal", + }, + }), + /single path segment/u, + ); +}); + +test("ambiguous response is captured only when its candidate observation receipt is readable", async () => { + const root = await mkdtemp(join(tmpdir(), "loopx-local-authority-shadow-")); + tCleanup(root); + class AfterCommitStore extends FileAuthorityStore { + protected override async replaceDurably(path: string, payload: Uint8Array): Promise { + await super.replaceDurably(path, payload); + if (path === this.path) throw new Error("lost response after durable replace"); + } + } + const recovered = await recordLocalAuthorityShadow(request(root), { + openStore: (directory, goalId) => new AfterCommitStore(directory, goalId), + }); + assert.equal(recovered.outcome, "ambiguous_reconciled"); + + const beforeRoot = await mkdtemp(join(tmpdir(), "loopx-local-authority-shadow-")); + tCleanup(beforeRoot); + class BeforeCommitStore extends FileAuthorityStore { + protected override async replaceDurably(path: string, _payload: Uint8Array): Promise { + if (path === this.path) throw new Error("failed before durable replace"); + return await super.replaceDurably(path, _payload); + } + } + const unproved = await recordLocalAuthorityShadow(request(beforeRoot), { + openStore: (directory, goalId) => new BeforeCommitStore(directory, goalId), + }); + assert.equal(unproved.outcome, "ambiguous_unproved"); +}); + +const cleanupRoots: string[] = []; +function tCleanup(root: string): void { + cleanupRoots.push(root); +} +test.after(async () => { + await Promise.all(cleanupRoots.map((root) => rm(root, { recursive: true, force: true }))); +}); diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index ce6722cb12..427768a5d6 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -27,6 +27,7 @@ "loopx/control_plane/coordination/runtime_shadow.ts", "loopx/control_plane/coordination/nokv_authority_store.ts", "loopx/control_plane/coordination/nokv_jsonl_transport.ts", + "loopx/control_plane/coordination/local_authority_shadow.ts", "loopx/control_plane/coordination/postgresql_authority_store.ts", "loopx/control_plane/agents/delivery_workspace.ts", "loopx/control_plane/goals/vision_checkpoint.ts", @@ -61,6 +62,7 @@ "tests/control_plane_ts/interaction_contract.test.ts", "tests/control_plane_ts/runtime_decode.test.ts", "tests/control_plane_ts/authority_store.test.ts", + "tests/control_plane_ts/local_authority_shadow.test.ts", "tests/control_plane_ts/coordination_runtime_shadow.test.ts", "tests/control_plane_ts/coordination_projection.test.ts", "tests/control_plane_ts/local_authority_runtime.test.ts",