From 2965af6096e427ef71c75b77285e12c675fd5bf6 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 9 Sep 2026 13:48:14 +0800 Subject: [PATCH 1/3] fix(delivery): share typed claim diagnostics across writers and history Signed-off-by: huangruiteng --- .../typescript-control-plane-migration-v0.md | 8 ++++ ...script-control-plane-migration-v0.zh-CN.md | 5 +++ .../control_plane/effect_runtime_handlers.ts | 2 + .../work_items/delivery_history.py | 13 +++++- .../work_items/delivery_history.ts | 8 +++- .../work_items/delivery_outcome.ts | 31 +++++++++++++ loopx/history.py | 4 ++ loopx/state_refresh.py | 8 ++++ loopx/status.py | 2 + .../test_delivery_claim_validation.py | 43 +++++++++++++++++++ .../control_plane/test_delivery_semantics.py | 10 +++-- .../test_delivery_semantics_cli.py | 18 ++++++++ .../control_plane_ts/delivery_history.test.ts | 27 ++++++++++++ 13 files changed, 174 insertions(+), 5 deletions(-) create mode 100644 tests/control_plane/test_delivery_claim_validation.py diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 9594ff5f0a..69b0d44190 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -182,6 +182,14 @@ label; no legacy prediction is retained without a concrete display consumer. State-only refresh remains legal without a delivery claim; this patch does not require every status refresh to declare progress. Existing write-time enum rejection, settlement evidence, quota, and gate checks remain in force. +- Follow-up claim validation shares one TS diagnosis with historical projection. + New writes reject progress paired with preparation-only work, primary outcome + paired with blocker work/typed blocked observation, and primary outcome paired + with an explicit follow-through requirement. Historical conflicts remain + readable as `unknown` plus `delivery_claim_conflicts`; persisted records and + settled receipts are never rewritten. Partial progress and valid blocker + writebacks remain legal. This is an intentional authoring/readback behavior + correction, not a change to small-delivery policy or a new evidence validator. - Legacy outcome-marker/hint configuration remains readable and preserves whether an outcome floor is configured. Its words no longer classify runs. No persisted history is rewritten and no new default-off flag restores the diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index dc133c253c..9fcb1cb0c0 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -141,6 +141,11 @@ classification 保留为历史标签；没有明确展示消费者时，不保 - 历史字段缺失或不受支持时保持 unknown。unknown 中断连续小规模／outcome-gap 证据计数，不视为成功或推断出的失败。未配置 floor 且没有 outcome 时，保留 `not_configured` 展示哨兵值。 +- 后续声明校验与历史投影共用一份 TS 诊断：新写入拒绝“进展 + 仅准备”、 + “主要成果 + blocker／typed blocked observation”和“主要成果 + 显式继续义务” + 的矛盾组合。历史冲突以 `unknown` 和 `delivery_claim_conflicts` 保持可读， + 不改写持久化记录或已结算 receipt；合法局部进展和 blocker 写回不受影响。 + 这是有意的写入／读取行为修正，不调整小步交付策略，也不是新的证据验证器。 - 新交付声明通过现有 writer API 写显式 enum，例如 `refresh-state --delivery-outcome ... --delivery-batch-scale ...`。 纯状态刷新仍可不声明交付；本批不强迫每次刷新声明进展。既有写入 enum 校验、 diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 2505f99c73..e94029f924 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -96,6 +96,7 @@ import { } from "./work_items/action_portfolio.ts"; import { projectQuotaPlanningHorizon } from "./work_items/planning_horizon.ts"; import { projectDeliveryHistory } from "./work_items/delivery_history.ts"; +import { validateDeliveryClaim } from "./work_items/delivery_outcome.ts"; import { evaluateTaskLeaseAcquireDecision, evaluateTaskLeaseWriteScopesOverlap, @@ -410,6 +411,7 @@ export function createEffectRuntimeHandlers( ["work_item.planning_inventory.detail", projectTodoPlanningInventoryDetail], ["work_item.refresh_recommendation.resolve", resolveRefreshRecommendation], ["work_item.delivery_history.project", projectDeliveryHistory], + ["work_item.delivery_claim.validate", validateDeliveryClaim], ["goal.vision_checkpoint.evaluate", buildVisionCheckpoint], ["goal.vision_wait.coverage", projectVisionWaitCoverage], ["goal.shared_goal_alignment.project", projectSharedGoalAlignment], diff --git a/loopx/control_plane/work_items/delivery_history.py b/loopx/control_plane/work_items/delivery_history.py index 09c430a041..40960b3157 100644 --- a/loopx/control_plane/work_items/delivery_history.py +++ b/loopx/control_plane/work_items/delivery_history.py @@ -1,4 +1,4 @@ -"""Compact transport for the typed delivery-history read model, never a writer.""" +"""Compact transport for typed delivery diagnostics and pre-write validation.""" from __future__ import annotations from collections.abc import Mapping @@ -70,3 +70,14 @@ def project_delivery_history( # Display-only annotation after the decision; narrative never enters TS. hint["latest_classification"] = _text(run.get("classification")).strip() return result + + +def require_consistent_delivery_claim(record: Mapping[str, Any]) -> None: + """Reject contradictory authored claims before effects; add no new fields.""" + if not any(record.get(key) for key in ("delivery_outcome", "delivery_turn_kind", "outcome_followthrough_required")): + return + result = effect_runtime_result("work_item.delivery_claim.validate", _run_facts(record)) + if not isinstance(result, dict) or result.get("schema_version") != "delivery_claim_validation_v0": + raise RuntimeError("TypeScript delivery claim validation shape mismatch") + if result.get("valid") is not True: + raise ValueError("contradictory delivery claim: " + ", ".join(result.get("conflicts") or [])) diff --git a/loopx/control_plane/work_items/delivery_history.ts b/loopx/control_plane/work_items/delivery_history.ts index f211b76633..ade5581188 100644 --- a/loopx/control_plane/work_items/delivery_history.ts +++ b/loopx/control_plane/work_items/delivery_history.ts @@ -1,7 +1,7 @@ import type { JsonObject } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; import { requireJsonObject } from "../runtime_decode.ts"; -import { DELIVERY_OUTCOMES, isTurnScopedSettlementOutcome, type DeliveryOutcome } from "./delivery_outcome.ts"; +import { DELIVERY_OUTCOMES, diagnoseDeliveryClaim, isTurnScopedSettlementOutcome, type DeliveryOutcome } from "./delivery_outcome.ts"; const TURN_KINDS = [ "contract_only_preparation", "compact_evidence", "blocker_writeback", @@ -117,6 +117,12 @@ export function projectDeliveryHistory(value: unknown): JsonObject { } const floorConfigured = input.outcome_floor_configured; const runs: DeliverySignal[] = input.runs.map(decodeRun).map((run) => { + const conflicts = diagnoseDeliveryClaim({ ...run }); + if (conflicts.length > 0) return { + delivery_outcome: "unknown", delivery_batch_scale: batchScale(run.delivery_batch_scale), + delivery_turn_kind: "unknown", outcome_followthrough: null, + delivery_claim_conflicts: conflicts, + }; const outcome = outcomeSignal(run.delivery_outcome, floorConfigured); const kind = turnKind(run, outcome); return { diff --git a/loopx/control_plane/work_items/delivery_outcome.ts b/loopx/control_plane/work_items/delivery_outcome.ts index b1f003b053..00e72d3c5b 100644 --- a/loopx/control_plane/work_items/delivery_outcome.ts +++ b/loopx/control_plane/work_items/delivery_outcome.ts @@ -1,4 +1,6 @@ import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { requireJsonObject } from "../runtime_decode.ts"; +import type { JsonObject } from "../effect_program.ts"; export const DELIVERY_OUTCOMES = [ "surface_only", @@ -17,6 +19,35 @@ export type DeliveryOutcome = (typeof DELIVERY_OUTCOMES)[number]; export type MaterialDeliveryOutcome = (typeof MATERIAL_DELIVERY_OUTCOMES)[number]; +export type DeliveryClaimConflict = + | "progress_with_preparation_only" + | "primary_outcome_with_blocker" + | "primary_outcome_with_followthrough"; + +/** Validate combinations, not prose or evidence truth. Historical readers use + * these same codes without rewriting already committed records or receipts. */ +export function diagnoseDeliveryClaim(raw: JsonObject): DeliveryClaimConflict[] { + const outcome = String(raw.delivery_outcome ?? "").trim(); + const kind = String(raw.delivery_turn_kind ?? "").trim(); + const conflicts: DeliveryClaimConflict[] = []; + if ((outcome === "outcome_progress" || outcome === "primary_goal_outcome") + && kind === "contract_only_preparation") conflicts.push("progress_with_preparation_only"); + const observation = jsonObject(raw.progress_observation); + if (outcome === "primary_goal_outcome" && (kind === "blocker_writeback" + || (observation?.schema_version === "typed_progress_observation_v0" && observation.result_class === "blocked"))) { + conflicts.push("primary_outcome_with_blocker"); + } + if (outcome === "primary_goal_outcome" && raw.outcome_followthrough_required === true) { + conflicts.push("primary_outcome_with_followthrough"); + } + return conflicts; +} + +export function validateDeliveryClaim(value: unknown): JsonObject { + const conflicts = diagnoseDeliveryClaim(requireJsonObject(value, "delivery claim")); + return { schema_version: "delivery_claim_validation_v0", valid: conflicts.length === 0, conflicts }; +} + const STABLE_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$/; const PROGRESS_DELIVERY_OUTCOMES = new Set([ "outcome_progress", diff --git a/loopx/history.py b/loopx/history.py index 306523dcd9..26211faa12 100644 --- a/loopx/history.py +++ b/loopx/history.py @@ -110,6 +110,10 @@ def write_reserved_run_artifacts( render_markdown: Callable[[dict[str, Any]], str], ) -> None: from .control_plane.quota.usage_collector import ingest_usage_into_run_record + from .control_plane.work_items.delivery_history import require_consistent_delivery_claim + + require_consistent_delivery_claim(record) + require_consistent_delivery_claim(index_record) # Producer call site for GH-C95: normalize any compact usage measurement onto # the durable run record and its index row before append. Fail closed here so diff --git a/loopx/state_refresh.py b/loopx/state_refresh.py index a9060498c3..5348ebc0ca 100644 --- a/loopx/state_refresh.py +++ b/loopx/state_refresh.py @@ -10,6 +10,7 @@ from typing import Any from .control_plane.runtime.time import now_local_iso +from .control_plane.work_items.delivery_history import require_consistent_delivery_claim from .control_plane.work_items.delivery_batch_scale import ( DELIVERY_BATCH_SCALE_CHOICES as DELIVERY_BATCH_SCALE_CHOICES, require_delivery_batch_scale, @@ -122,6 +123,13 @@ def _serialized_refresh( @wraps(function) def run(**kwargs: Any) -> dict[str, Any]: goal_id = validate_goal_id_path_segment(kwargs["goal_id"]) + observation = kwargs.get("progress_observation") + require_consistent_delivery_claim({ + "delivery_outcome": kwargs.get("delivery_outcome"), + "progress_observation": normalize_progress_observation( + observation, work_item_id=kwargs.get("todo_id") or kwargs.get("replan_obligation_id"), + ) if observation is not None else None, + }) registry = load_registry(kwargs["registry_path"]) root = resolve_runtime_root(registry, kwargs["runtime_root_override"]) if kwargs["dry_run"]: diff --git a/loopx/status.py b/loopx/status.py index 40a39200d1..b789faa0a9 100644 --- a/loopx/status.py +++ b/loopx/status.py @@ -586,6 +586,8 @@ def project_post_handoff_history( "generated_at", "classification", "health_check", "json_exists", "markdown_exists", ) if field in run} compact.update({field: signal[field] for field in ("delivery_batch_scale", "delivery_turn_kind")}) + if signal.get("delivery_claim_conflicts"): + compact["delivery_claim_conflicts"] = signal["delivery_claim_conflicts"] if signal["delivery_outcome"] != "not_configured": compact["delivery_outcome"] = signal["delivery_outcome"] compact_runs.append(_attach_run_summary_projections_read_model( diff --git a/tests/control_plane/test_delivery_claim_validation.py b/tests/control_plane/test_delivery_claim_validation.py new file mode 100644 index 0000000000..d100fb0d7a --- /dev/null +++ b/tests/control_plane/test_delivery_claim_validation.py @@ -0,0 +1,43 @@ +"""Writer rejection and tolerant historical readback use the same TS rules.""" +from copy import deepcopy + +import pytest + +from loopx.control_plane.work_items.delivery_history import require_consistent_delivery_claim +from loopx.history import write_reserved_run_artifacts +from loopx.state_refresh import refresh_state_run +from loopx.status import compact_post_handoff_run + + +@pytest.mark.parametrize("record", [ + {"delivery_outcome": "outcome_progress", "delivery_turn_kind": "contract_only_preparation"}, + {"delivery_outcome": "primary_goal_outcome", "delivery_turn_kind": "blocker_writeback"}, + {"delivery_outcome": "primary_goal_outcome", "outcome_followthrough_required": True}, +]) +def test_historical_conflict_is_visible_and_new_write_has_no_artifacts(tmp_path, record): + before = deepcopy(record) + compact = compact_post_handoff_run(record) + assert compact["delivery_outcome"] == "unknown" + assert compact["delivery_claim_conflicts"] + with pytest.raises(ValueError, match="contradictory delivery claim"): + write_reserved_run_artifacts(runs_dir=tmp_path, generated_at="2026-09-09T00:00:00Z", + record=record, index_record={}, payload={}, render_markdown=lambda _: "unused") + assert list(tmp_path.iterdir()) == [] + assert record == before + + +def test_refresh_rejects_primary_blocker_before_loading_registry_or_mutating_state(tmp_path): + with pytest.raises(ValueError, match="primary_outcome_with_blocker"): + refresh_state_run(registry_path=tmp_path / "absent-registry.json", goal_id="delivery", + classification="synthetic_delivery", delivery_outcome="primary_goal_outcome", + todo_id="todo_delivery", progress_observation={ + "schema_version": "typed_progress_observation_v0", "result_class": "blocked", + "work_item_id": "todo_delivery", "blocker_id": "blocker-a", "evidence_ids": ["evidence-a"], + }) + assert list(tmp_path.iterdir()) == [] + + +def test_valid_partial_progress_and_unspecified_delivery_remain_legal(): + for record in ({}, {"delivery_outcome": "outcome_progress", "delivery_turn_kind": "product_path_execution"}, + {"delivery_outcome": "outcome_gap", "delivery_turn_kind": "blocker_writeback"}): + require_consistent_delivery_claim(record) diff --git a/tests/control_plane/test_delivery_semantics.py b/tests/control_plane/test_delivery_semantics.py index f2cbd433f6..3dfaf6b28c 100644 --- a/tests/control_plane/test_delivery_semantics.py +++ b/tests/control_plane/test_delivery_semantics.py @@ -163,10 +163,14 @@ def test_followthrough_requires_literal_true_not_truthy_metadata(required) -> No assert build_outcome_followthrough_hint({"outcome_followthrough_required": required}) is None -def test_primary_outcome_precedes_explicit_followthrough_and_unknown_kind_does_not_recover() -> None: - assert build_outcome_followthrough_hint({ +def test_contradictory_primary_claim_is_diagnostic_and_unknown_kind_does_not_recover() -> None: + contradictory = { "delivery_outcome": "primary_goal_outcome", "outcome_followthrough_required": True, - }) is None + } + assert build_outcome_followthrough_hint(contradictory) is None + compact = compact_post_handoff_run(contradictory) + assert compact["delivery_outcome"] == "unknown" + assert compact["delivery_claim_conflicts"] == ["primary_outcome_with_followthrough"] assert delivery_turn_kind_for_run({ "delivery_outcome": "primary_goal_outcome", "delivery_turn_kind": "future_kind", }) == "unknown" diff --git a/tests/control_plane/test_delivery_semantics_cli.py b/tests/control_plane/test_delivery_semantics_cli.py index fb6a81e621..d883a4d202 100644 --- a/tests/control_plane/test_delivery_semantics_cli.py +++ b/tests/control_plane/test_delivery_semantics_cli.py @@ -76,3 +76,21 @@ def cli(*args: str) -> dict: else: assert compact["delivery_turn_kind"] == "compact_evidence" assert build_outcome_followthrough_hint(compact) is None + + # The real authoring entrypoint rejects a contradictory claim before any + # state/history changes, including in dry-run mode. + before = {path: path.read_bytes() for path in tmp_path.rglob("*") if path.is_file()} + for mode in ([], ["--dry-run"]): + rejected = subprocess.run( + [sys.executable, "-m", "loopx.cli", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json", "refresh-state", + "--goal-id", "delivery-semantics", "--classification", "typed claim", + "--delivery-outcome", "primary_goal_outcome", "--todo-id", "todo_delivery", + "--progress-result-class", "blocked", "--progress-blocker-id", "blocker-a", + "--progress-evidence-id", "evidence-a", "--no-global-sync", *mode], + cwd=tmp_path, env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, text=True, timeout=30, check=False, + ) + assert rejected.returncode != 0 + assert "primary_outcome_with_blocker" in rejected.stdout + rejected.stderr + assert {path: path.read_bytes() for path in tmp_path.rglob("*") if path.is_file()} == before diff --git a/tests/control_plane_ts/delivery_history.test.ts b/tests/control_plane_ts/delivery_history.test.ts index 3e43f5da47..f5d0586f04 100644 --- a/tests/control_plane_ts/delivery_history.test.ts +++ b/tests/control_plane_ts/delivery_history.test.ts @@ -2,6 +2,7 @@ import assert from "node:assert/strict"; import test from "node:test"; import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; import { projectDeliveryHistory } from "../../loopx/control_plane/work_items/delivery_history.ts"; +import { validateDeliveryClaim } from "../../loopx/control_plane/work_items/delivery_outcome.ts"; function run(fields: JsonObject = {}): JsonObject { return { delivery_outcome: "", delivery_batch_scale: "", delivery_turn_kind: "", @@ -101,3 +102,29 @@ test("wire faults fail closed, not a fallback to untyped history", () => { assert.throws(() => project(rows)); } }); + +test("contradictory historical declarations are diagnostic, not progress or new obligations", () => { + for (const fields of [ + { delivery_outcome: "outcome_progress", delivery_turn_kind: "contract_only_preparation" }, + { delivery_outcome: "primary_goal_outcome", delivery_turn_kind: "blocker_writeback" }, + { delivery_outcome: "primary_goal_outcome", outcome_followthrough_required: true }, + { delivery_outcome: "primary_goal_outcome", progress_observation: { + schema_version: "typed_progress_observation_v0", result_class: "blocked" } }, + ]) { + const record = run(fields); + const before = structuredClone(record); + assert.equal(validateDeliveryClaim(record).valid, false); + const projected = signal(record); + assert.equal(projected.delivery_outcome, "unknown"); + assert.equal(projected.delivery_turn_kind, "unknown"); + assert.equal(projected.outcome_followthrough, null); + assert.deepEqual(projected.delivery_claim_conflicts, validateDeliveryClaim(record).conflicts); + assert.deepEqual(record, before); + } + for (const fields of [ + { delivery_outcome: "outcome_progress", delivery_turn_kind: "product_path_execution" }, + { delivery_outcome: "primary_goal_outcome", delivery_turn_kind: "compact_evidence" }, + { delivery_outcome: "outcome_gap", delivery_turn_kind: "blocker_writeback" }, + {}, + ]) assert.equal(validateDeliveryClaim(run(fields)).valid, true); +}); From f23f2429464a931905670620108ebf60732b97f4 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 9 Sep 2026 17:00:24 +0800 Subject: [PATCH 2/3] fix(refresh): validate normalized claims before serialized effects Signed-off-by: huangruiteng --- .../typescript-control-plane-migration-v0.md | 4 + ...script-control-plane-migration-v0.zh-CN.md | 3 + loopx/state_refresh.py | 1180 ++++++++--------- .../test_delivery_claim_validation.py | 25 + 4 files changed, 611 insertions(+), 601 deletions(-) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 69b0d44190..29ae29a956 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -190,6 +190,10 @@ label; no legacy prediction is retained without a concrete display consumer. settled receipts are never rewritten. Partial progress and valid blocker writebacks remain legal. This is an intentional authoring/readback behavior correction, not a change to small-delivery policy or a new evidence validator. + Refresh validates individual fields in their established order, then checks + the normalized claim before registry access or lock creation. Invalid input + therefore takes precedence over store errors, including in dry-run mode; + state-dependent admission and writeback still share the same runtime lock. - Legacy outcome-marker/hint configuration remains readable and preserves whether an outcome floor is configured. Its words no longer classify runs. No persisted history is rewritten and no new default-off flag restores the diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 9fcb1cb0c0..9ab7952d7b 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -146,6 +146,9 @@ classification 保留为历史标签；没有明确展示消费者时，不保 的矛盾组合。历史冲突以 `unknown` 和 `delivery_claim_conflicts` 保持可读， 不改写持久化记录或已结算 receipt；合法局部进展和 blocker 写回不受影响。 这是有意的写入／读取行为修正，不调整小步交付策略，也不是新的证据验证器。 + Refresh 先按既有顺序校验各字段，再以归一化结果检查组合语义，之后才读取 + registry 和创建锁。因此非法输入优先于存储错误返回，dry-run 也一致；依赖 + 当前状态的准入与写回仍在同一 runtime 锁内完成。 - 新交付声明通过现有 writer API 写显式 enum，例如 `refresh-state --delivery-outcome ... --delivery-batch-scale ...`。 纯状态刷新仍可不声明交付；本批不强迫每次刷新声明进展。既有写入 enum 校验、 diff --git a/loopx/state_refresh.py b/loopx/state_refresh.py index 5348ebc0ca..ce58b9e068 100644 --- a/loopx/state_refresh.py +++ b/loopx/state_refresh.py @@ -3,9 +3,7 @@ import hashlib import json import re -from collections.abc import Callable -from functools import wraps -from contextlib import ExitStack +from contextlib import ExitStack, nullcontext from pathlib import Path from typing import Any @@ -111,37 +109,6 @@ REPAIR_NOOP_SCHEMA_VERSION = "repair_noop_v0" -def _serialized_refresh( - function: Callable[..., dict[str, Any]], -) -> Callable[..., dict[str, Any]]: - """Keep admission and the legacy index append in one cross-writer lock. - - Transitional Python persistence adapter; remove with the native refresh - writer. No external provider runs while this lock is held. - """ - - @wraps(function) - def run(**kwargs: Any) -> dict[str, Any]: - goal_id = validate_goal_id_path_segment(kwargs["goal_id"]) - observation = kwargs.get("progress_observation") - require_consistent_delivery_claim({ - "delivery_outcome": kwargs.get("delivery_outcome"), - "progress_observation": normalize_progress_observation( - observation, work_item_id=kwargs.get("todo_id") or kwargs.get("replan_obligation_id"), - ) if observation is not None else None, - }) - registry = load_registry(kwargs["registry_path"]) - root = resolve_runtime_root(registry, kwargs["runtime_root_override"]) - if kwargs["dry_run"]: - return function(**kwargs) - with exclusive_file_lock( - root / "goals" / goal_id / "runs" / "index.jsonl", operation="refresh-state" - ): - return function(**kwargs) - - return run - - def now_local() -> str: return now_local_iso() @@ -793,7 +760,6 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str: return "\n".join(lines) -@_serialized_refresh def refresh_state_run( *, registry_path: Path, @@ -862,6 +828,13 @@ def refresh_state_run( if progress_observation is not None else None ) + # Validate individual fields before asking the typed owner about their + # combination. Reuse normalized facts; malformed input must not open a + # registry or create a write lock in either normal or dry-run execution. + require_consistent_delivery_claim({ + "delivery_outcome": normalized_delivery_outcome, + "progress_observation": normalized_progress_observation, + }) turn_scoped_settlement_qualified = qualifies_turn_scoped_settlement( normalized_delivery_outcome, normalized_progress_observation, @@ -875,622 +848,627 @@ def refresh_state_run( ) registry = load_registry(registry_path) runtime_root = resolve_runtime_root(registry, runtime_root_override, registry_path=registry_path) - settlement_identity = None - settlement_result = None - delivery_workspace_causality = None - settlement_workspace_requirement = None - settlement_readback = None - refresh_recovery = None - prior_writeback_run = None - if todo_id or normalized_replan_obligation_id or turn_instance_id: - if not turn_scoped_settlement_qualified: - raise ValueError( - "turn-scoped refresh-state requires a progress outcome or a typed " - "blocked outcome_gap settlement" + # State-dependent admission through the final append remains serialized. + # Only pure input validation runs before this transitional persistence lock. + with (nullcontext() if dry_run else exclusive_file_lock( + runtime_root / "goals" / safe_goal_id / "runs" / "index.jsonl", operation="refresh-state" + )): + settlement_identity = None + settlement_result = None + delivery_workspace_causality = None + settlement_workspace_requirement = None + settlement_readback = None + refresh_recovery = None + prior_writeback_run = None + if todo_id or normalized_replan_obligation_id or turn_instance_id: + if not turn_scoped_settlement_qualified: + raise ValueError( + "turn-scoped refresh-state requires a progress outcome or a typed " + "blocked outcome_gap settlement" + ) + settlement_readback = read_heartbeat_settlement( + runtime_root, + goal_id=safe_goal_id, + agent_id=normalized_agent_id or None, + todo_id=todo_id, + turn_instance_id=turn_instance_id, + replan_obligation_id=normalized_replan_obligation_id, + refresh_retry={ + "vision": agent_vision_packet, + "unchanged_reason": vision_unchanged_reason, + "merge_patch": bool(merge_agent_vision_patch), + "workspace_requested": delivery_workspace_path is not None, + "mutation": { + "next_action": next_action, + "autonomous_replan_recorded": autonomous_replan_recorded, + "repair_delta_kinds": repair_delta_kinds, + "usage_measurement": usage_measurement, + "usage_codex_session": str(usage_codex_session) + if usage_codex_session + else None, + }, + "delivery_outcome": normalized_delivery_outcome, + "delivery_batch_scale": normalized_delivery_batch_scale, + "delivery_boundary": delivery_boundary, + "progress_observation": normalized_progress_observation, + }, ) - settlement_readback = read_heartbeat_settlement( - runtime_root, + if settlement_readback is None: + raise RuntimeError("exact settlement readback unexpectedly returned not-found") + settlement_result = settlement_readback.identity + if settlement_result.failure is not None: + raise ValueError(settlement_result.failure.reason) + settlement_identity = settlement_result.value + if settlement_identity is None: + raise ValueError("turn-scoped refresh-state has no settlement identity") + delivery_workspace_causality = settlement_readback.workspace_causality + refresh_recovery = settlement_readback.refresh_recovery + if not refresh_recovery: + raise RuntimeError("settlement readback omitted refresh recovery admission") + decision = refresh_recovery["decision"] + prior_writeback_run = settlement_readback.writeback_run + if decision in {"replay", "repair_receipt", "reject"}: + payload = { + **(prior_writeback_run or {}), + "ok": decision != "reject", + "dry_run": dry_run, + "appended": False, + "idempotent_replay": decision == "replay", + "receipt_repair_required": decision == "repair_receipt" and not dry_run, + "registry": str(registry_path), + "runtime_root": str(runtime_root), + "goal_id": safe_goal_id, + "refresh_recovery": refresh_recovery, + "settlement_identity": settlement_identity.as_dict(), + "settlement_result": settlement_result_payload( + settlement_readback.delivery + ), + } + if decision == "reject": + payload["error"] = ( + f"{refresh_recovery['reason']}: committed writeback is unchanged; " + "do not begin a new Turn or repeat spend to repair it. " + "Retry the original delivery fields with only the missing vision decision; " + "if a newer vision already exists, inspect current quota instead." + ) + return payload + settlement_workspace_requirement = resolve_settlement_workspace_requirement( + delivery_workspace_causality, settlement_binding_kind=settlement_identity.binding_kind.value + ) + runtime_projection_route = resolve_runtime_projection_route( + registry_path=registry_path, goal_id=safe_goal_id, - agent_id=normalized_agent_id or None, - todo_id=todo_id, - turn_instance_id=turn_instance_id, - replan_obligation_id=normalized_replan_obligation_id, - refresh_retry={ - "vision": agent_vision_packet, - "unchanged_reason": vision_unchanged_reason, - "merge_patch": bool(merge_agent_vision_patch), - "workspace_requested": delivery_workspace_path is not None, - "mutation": { - "next_action": next_action, - "autonomous_replan_recorded": autonomous_replan_recorded, - "repair_delta_kinds": repair_delta_kinds, - "usage_measurement": usage_measurement, - "usage_codex_session": str(usage_codex_session) - if usage_codex_session - else None, - }, - "delivery_outcome": normalized_delivery_outcome, - "delivery_batch_scale": normalized_delivery_batch_scale, - "delivery_boundary": delivery_boundary, - "progress_observation": normalized_progress_observation, - }, + source_runtime_root=runtime_root, ) - if settlement_readback is None: - raise RuntimeError("exact settlement readback unexpectedly returned not-found") - settlement_result = settlement_readback.identity - if settlement_result.failure is not None: - raise ValueError(settlement_result.failure.reason) - settlement_identity = settlement_result.value - if settlement_identity is None: - raise ValueError("turn-scoped refresh-state has no settlement identity") - delivery_workspace_causality = settlement_readback.workspace_causality - refresh_recovery = settlement_readback.refresh_recovery - if not refresh_recovery: - raise RuntimeError("settlement readback omitted refresh recovery admission") - decision = refresh_recovery["decision"] - prior_writeback_run = settlement_readback.writeback_run - if decision in {"replay", "repair_receipt", "reject"}: - payload = { - **(prior_writeback_run or {}), - "ok": decision != "reject", - "dry_run": dry_run, - "appended": False, - "idempotent_replay": decision == "replay", - "receipt_repair_required": decision == "repair_receipt" and not dry_run, - "registry": str(registry_path), - "runtime_root": str(runtime_root), - "goal_id": safe_goal_id, - "refresh_recovery": refresh_recovery, - "settlement_identity": settlement_identity.as_dict(), - "settlement_result": settlement_result_payload( - settlement_readback.delivery - ), - } - if decision == "reject": - payload["error"] = ( - f"{refresh_recovery['reason']}: committed writeback is unchanged; " - "do not begin a new Turn or repeat spend to repair it. " - "Retry the original delivery fields with only the missing vision decision; " - "if a newer vision already exists, inspect current quota instead." - ) - return payload - settlement_workspace_requirement = resolve_settlement_workspace_requirement( - delivery_workspace_causality, settlement_binding_kind=settlement_identity.binding_kind.value + route_status = str(runtime_projection_route.get("status") or "missing") + route_target_text = str( + runtime_projection_route.get("target_runtime_root") or "" + ).strip() + route_target_root = Path(route_target_text) if route_target_text else None + shared_runtime_root = ( + route_target_root if sync_global and route_status == "resolved" else None ) - runtime_projection_route = resolve_runtime_projection_route( - registry_path=registry_path, - goal_id=safe_goal_id, - source_runtime_root=runtime_root, - ) - route_status = str(runtime_projection_route.get("status") or "missing") - route_target_text = str( - runtime_projection_route.get("target_runtime_root") or "" - ).strip() - route_target_root = Path(route_target_text) if route_target_text else None - shared_runtime_root = ( - route_target_root if sync_global and route_status == "resolved" else None - ) - global_sync_runtime_root = ( - route_target_root - if sync_global and route_status in {"resolved", "single_runtime"} - else None - ) - registry_goal, resolved_project, resolved_state_file = resolve_goal_state( - registry=registry, - goal_id=safe_goal_id, - project_override=project, - state_file_override=state_file, - ) - state_text, planning_events, todo_fields = load_refresh_planning_source( - runtime_root, safe_goal_id, resolved_state_file, require_display=bool(next_action) - ) - expected_write_state_text = state_text - if normalized_delivery_outcome in ACCOUNTABLE_DELIVERY_OUTCOMES: - require_accountable_completion_validation( - state_text, - todo_fields=todo_fields, - todo_id=(settlement_identity.todo_id if settlement_identity else None), - agent_id=normalized_agent_id or None, - ) - normalized_next_action = normalize_next_action_text(next_action) if next_action else None - registered_agents = registered_agents_for_goal(registry_goal) - known_agents = {agent for agent in registered_agents if agent} - multi_agent_goal = len(known_agents) > 1 - workspace_guard_policy = ( - registry_goal.get("workspace_guard_policy") - if isinstance(registry_goal.get("workspace_guard_policy"), dict) - else {} - ) - explicit_peer_worktree_requirement = workspace_guard_policy.get( - "peer_independent_worktree_required" - ) - peer_independent_worktree_required = multi_agent_goal and ( - explicit_peer_worktree_requirement is None - or explicit_peer_worktree_requirement is True - ) - if normalized_agent_id and known_agents and normalized_agent_id not in known_agents: - raise ValueError( - f"agent_id {normalized_agent_id!r} is not registered for goal {safe_goal_id!r}" + global_sync_runtime_root = ( + route_target_root + if sync_global and route_status in {"resolved", "single_runtime"} + else None ) - if multi_agent_goal and not normalized_agent_id: - raise ValueError( - "multi-agent refresh-state requires --agent-id; text inference is disabled" + registry_goal, resolved_project, resolved_state_file = resolve_goal_state( + registry=registry, + goal_id=safe_goal_id, + project_override=project, + state_file_override=state_file, ) - if not normalized_progress_scope: - normalized_progress_scope = ( - AGENT_LANE_PROGRESS_SCOPE if normalized_agent_id else GOAL_PROGRESS_SCOPE + state_text, planning_events, todo_fields = load_refresh_planning_source( + runtime_root, safe_goal_id, resolved_state_file, require_display=bool(next_action) ) - if normalized_progress_scope == AGENT_LANE_PROGRESS_SCOPE: - if not normalized_agent_id: - raise ValueError("--progress-scope agent_lane requires --agent-id") - if normalized_next_action: - raise ValueError( - "agent-lane refresh-state cannot update the durable active-state Next Action; " - "rerun without --next-action or use --progress-scope goal from a registered peer" + expected_write_state_text = state_text + if normalized_delivery_outcome in ACCOUNTABLE_DELIVERY_OUTCOMES: + require_accountable_completion_validation( + state_text, + todo_fields=todo_fields, + todo_id=(settlement_identity.todo_id if settlement_identity else None), + agent_id=normalized_agent_id or None, ) - if normalized_progress_scope == GOAL_PROGRESS_SCOPE: - if normalized_agent_lane: - raise ValueError("--agent-lane requires --progress-scope agent_lane") - if (agent_vision_packet is not None or vision_unchanged_reason) and not normalized_agent_id: - raise ValueError("vision writeback requires --agent-id") - agent_vision: dict[str, Any] | None = None - existing_agent_vision: dict[str, Any] | None = None - autonomous_replan_frontier_identity: str | None = None - newest_first_runs: list[dict[str, Any]] = [] - if normalized_agent_id: - existing_runs, _ = load_index( - runtime_root / "goals" / safe_goal_id / "runs" / "index.jsonl" + normalized_next_action = normalize_next_action_text(next_action) if next_action else None + registered_agents = registered_agents_for_goal(registry_goal) + known_agents = {agent for agent in registered_agents if agent} + multi_agent_goal = len(known_agents) > 1 + workspace_guard_policy = ( + registry_goal.get("workspace_guard_policy") + if isinstance(registry_goal.get("workspace_guard_policy"), dict) + else {} ) - newest_first_runs = [ - run - for _, run in sorted( - enumerate(existing_runs), - key=lambda item: (str(item[1].get("generated_at") or ""), item[0]), - reverse=True, - ) - ] - existing_agent_vision = latest_agent_vision_from_runs( - newest_first_runs, - goal_id=safe_goal_id, - agent_id=normalized_agent_id, + explicit_peer_worktree_requirement = workspace_guard_policy.get( + "peer_independent_worktree_required" ) - if agent_vision_packet is not None: - agent_vision = prepare_vision_refresh( - agent_vision_packet, - goal_id=safe_goal_id, - agent_id=normalized_agent_id or None, - existing_agent_vision=existing_agent_vision, - merge_patch=merge_agent_vision_patch, - require_path_delta_for_durable_change=autonomous_replan_recorded, + peer_independent_worktree_required = multi_agent_goal and ( + explicit_peer_worktree_requirement is None + or explicit_peer_worktree_requirement is True ) - generated_at = now_local() - active_state_next_action_update: dict[str, Any] | None = None - if normalized_next_action: - with exclusive_cross_runtime_file_lock(resolved_state_file): - locked_state_text = resolved_state_file.read_text(encoding="utf-8") - expected_write_state_text = locked_state_text - updated_state_text, state_updated = replace_next_action_section( - locked_state_text, - next_action=normalized_next_action, - updated_at=generated_at, + if normalized_agent_id and known_agents and normalized_agent_id not in known_agents: + raise ValueError( + f"agent_id {normalized_agent_id!r} is not registered for goal {safe_goal_id!r}" ) - active_state_next_action_update = { - "schema_version": ACTIVE_STATE_NEXT_ACTION_UPDATE_SCHEMA_VERSION, - "source": "refresh_state", - "next_action": normalized_next_action, - "updated": bool(state_updated and not dry_run), - "would_update": bool(state_updated), - "dry_run": bool(dry_run), - "updated_at": generated_at if state_updated else None, - } - state_text = updated_state_text if state_updated else locked_state_text - - recommendation_resolution = resolve_refresh_recommendation( - state_text, - todo_fields=todo_fields, - explicit_action=recommended_action, - agent_id=normalized_agent_id or None, - settlement_identity=( - settlement_identity.as_dict() if settlement_identity is not None else None - ), - registry_goal=registry_goal, - state_path=resolved_state_file, - rollout_events=( - planning_events if normalized_agent_id or settlement_identity is not None else None - ), - ) - action = str(recommendation_resolution["recommended_action"]) - recommended_action_source = str( - recommendation_resolution["recommended_action_source"] - ) - requested_classification = classification - settlement_replan_guard = ( - settlement_readback.semantic_replan_guard - if settlement_readback is not None - and settlement_readback.semantic_replan_guard is not None - else {} - ) - replan_qualification = qualify_refresh_replan_writeback( - todo_fields=todo_fields, - autonomous_replan_recorded=autonomous_replan_recorded, - requested_delta_kinds=normalized_repair_delta_kinds, - active_state_next_action_update=active_state_next_action_update, - agent_vision=agent_vision, - existing_agent_vision=existing_agent_vision, - agent_id=normalized_agent_id, - dry_run=dry_run, - settlement_todo_id=(settlement_identity.todo_id if settlement_identity else None), - settlement_guard_scoped=( - settlement_replan_guard.get("scope") == "turn_guard" - ), - settlement_guard_semantic_replan_obligation_id=( - settlement_replan_guard.get("selected_obligation_id") - if isinstance( - settlement_replan_guard.get("selected_obligation_id"), str + if multi_agent_goal and not normalized_agent_id: + raise ValueError( + "multi-agent refresh-state requires --agent-id; text inference is disabled" ) - else None - ), - newest_first_runs=newest_first_runs, - state_text=state_text, - goal_id=safe_goal_id, - progress_observation=normalized_progress_observation, - registry_goal=registry_goal, - completion_todo_id=completion_todo_id, - completion_turn_key=completion_turn_key, - classification=classification, - delivery_outcome=normalized_delivery_outcome, - ) - repair_delta_contract = replan_qualification.repair_delta_contract - replan_semantic_delta = replan_qualification.semantic_delta - autonomous_replan_frontier_identity = replan_qualification.frontier_identity - classification = replan_qualification.classification - normalized_delivery_outcome = replan_qualification.delivery_outcome - effective_autonomous_replan_recorded = ( - replan_qualification.autonomous_replan_recorded - ) - vision_checkpoint = build_vision_checkpoint( - agent_id=normalized_agent_id or None, - agent_vision=agent_vision, - existing_agent_vision=existing_agent_vision, - vision_unchanged_reason=vision_unchanged_reason, - delivery_outcome=normalized_delivery_outcome, - active_state_next_action_update=active_state_next_action_update, - delivery_boundary=delivery_boundary, - todo_id=(settlement_identity.todo_id if settlement_identity else None), - completion_todo_id=completion_todo_id, - autonomous_replan_recorded=effective_autonomous_replan_recorded, - ) - checkpoint_supplement = bool( - refresh_recovery and refresh_recovery["decision"] == "supplement_checkpoint" - ) - if checkpoint_supplement and not vision_checkpoint.get("satisfied"): - raise ValueError( - "checkpoint supplement did not satisfy the missing decision; " - "an unchanged reason requires an existing vision. Supply a valid vision " - "patch on the same Turn; the original writeback and quota are unchanged." + if not normalized_progress_scope: + normalized_progress_scope = ( + AGENT_LANE_PROGRESS_SCOPE if normalized_agent_id else GOAL_PROGRESS_SCOPE + ) + if normalized_progress_scope == AGENT_LANE_PROGRESS_SCOPE: + if not normalized_agent_id: + raise ValueError("--progress-scope agent_lane requires --agent-id") + if normalized_next_action: + raise ValueError( + "agent-lane refresh-state cannot update the durable active-state Next Action; " + "rerun without --next-action or use --progress-scope goal from a registered peer" + ) + if normalized_progress_scope == GOAL_PROGRESS_SCOPE: + if normalized_agent_lane: + raise ValueError("--agent-lane requires --progress-scope agent_lane") + if (agent_vision_packet is not None or vision_unchanged_reason) and not normalized_agent_id: + raise ValueError("vision writeback requires --agent-id") + agent_vision: dict[str, Any] | None = None + existing_agent_vision: dict[str, Any] | None = None + autonomous_replan_frontier_identity: str | None = None + newest_first_runs: list[dict[str, Any]] = [] + if normalized_agent_id: + existing_runs, _ = load_index( + runtime_root / "goals" / safe_goal_id / "runs" / "index.jsonl" + ) + newest_first_runs = [ + run + for _, run in sorted( + enumerate(existing_runs), + key=lambda item: (str(item[1].get("generated_at") or ""), item[0]), + reverse=True, + ) + ] + existing_agent_vision = latest_agent_vision_from_runs( + newest_first_runs, + goal_id=safe_goal_id, + agent_id=normalized_agent_id, + ) + if agent_vision_packet is not None: + agent_vision = prepare_vision_refresh( + agent_vision_packet, + goal_id=safe_goal_id, + agent_id=normalized_agent_id or None, + existing_agent_vision=existing_agent_vision, + merge_patch=merge_agent_vision_patch, + require_path_delta_for_durable_change=autonomous_replan_recorded, + ) + generated_at = now_local() + active_state_next_action_update: dict[str, Any] | None = None + if normalized_next_action: + with exclusive_cross_runtime_file_lock(resolved_state_file): + locked_state_text = resolved_state_file.read_text(encoding="utf-8") + expected_write_state_text = locked_state_text + updated_state_text, state_updated = replace_next_action_section( + locked_state_text, + next_action=normalized_next_action, + updated_at=generated_at, + ) + active_state_next_action_update = { + "schema_version": ACTIVE_STATE_NEXT_ACTION_UPDATE_SCHEMA_VERSION, + "source": "refresh_state", + "next_action": normalized_next_action, + "updated": bool(state_updated and not dry_run), + "would_update": bool(state_updated), + "dry_run": bool(dry_run), + "updated_at": generated_at if state_updated else None, + } + state_text = updated_state_text if state_updated else locked_state_text + + recommendation_resolution = resolve_refresh_recommendation( + state_text, + todo_fields=todo_fields, + explicit_action=recommended_action, + agent_id=normalized_agent_id or None, + settlement_identity=( + settlement_identity.as_dict() if settlement_identity is not None else None + ), + registry_goal=registry_goal, + state_path=resolved_state_file, + rollout_events=( + planning_events if normalized_agent_id or settlement_identity is not None else None + ), ) - delivery_workspace = None - workspace_requirement = str( - (settlement_workspace_requirement or {}).get("requirement") or "unknown" - ) - if delivery_workspace_path is not None and workspace_requirement == "not_required": - raise ValueError( - "--delivery-workspace-path conflicts with the original Todo's " - "explicit non-delivery settlement contract" + action = str(recommendation_resolution["recommended_action"]) + recommended_action_source = str( + recommendation_resolution["recommended_action_source"] ) - if ( - turn_scoped_settlement_qualified - and workspace_requirement != "not_required" - and not checkpoint_supplement - ): - delivery_workspace = capture_delivery_workspace( - current_path=delivery_workspace_path, - peer_independent_worktree_required=peer_independent_worktree_required, - local_goal_id=safe_goal_id, - local_project_root=resolved_project, - repository_source=( - "refresh_state.delivery_workspace_path" - if delivery_workspace_path is not None + requested_classification = classification + settlement_replan_guard = ( + settlement_readback.semantic_replan_guard + if settlement_readback is not None + and settlement_readback.semantic_replan_guard is not None + else {} + ) + replan_qualification = qualify_refresh_replan_writeback( + todo_fields=todo_fields, + autonomous_replan_recorded=autonomous_replan_recorded, + requested_delta_kinds=normalized_repair_delta_kinds, + active_state_next_action_update=active_state_next_action_update, + agent_vision=agent_vision, + existing_agent_vision=existing_agent_vision, + agent_id=normalized_agent_id, + dry_run=dry_run, + settlement_todo_id=(settlement_identity.todo_id if settlement_identity else None), + settlement_guard_scoped=( + settlement_replan_guard.get("scope") == "turn_guard" + ), + settlement_guard_semantic_replan_obligation_id=( + settlement_replan_guard.get("selected_obligation_id") + if isinstance( + settlement_replan_guard.get("selected_obligation_id"), str + ) else None ), + newest_first_runs=newest_first_runs, + state_text=state_text, + goal_id=safe_goal_id, + progress_observation=normalized_progress_observation, + registry_goal=registry_goal, + completion_todo_id=completion_todo_id, + completion_turn_key=completion_turn_key, + classification=classification, + delivery_outcome=normalized_delivery_outcome, ) - if ( - peer_independent_worktree_required - and ( - delivery_workspace is None - or delivery_workspace.get("workspace_kind") - != "independent_git_worktree" - ) - ): + repair_delta_contract = replan_qualification.repair_delta_contract + replan_semantic_delta = replan_qualification.semantic_delta + autonomous_replan_frontier_identity = replan_qualification.frontier_identity + classification = replan_qualification.classification + normalized_delivery_outcome = replan_qualification.delivery_outcome + effective_autonomous_replan_recorded = ( + replan_qualification.autonomous_replan_recorded + ) + vision_checkpoint = build_vision_checkpoint( + agent_id=normalized_agent_id or None, + agent_vision=agent_vision, + existing_agent_vision=existing_agent_vision, + vision_unchanged_reason=vision_unchanged_reason, + delivery_outcome=normalized_delivery_outcome, + active_state_next_action_update=active_state_next_action_update, + delivery_boundary=delivery_boundary, + todo_id=(settlement_identity.todo_id if settlement_identity else None), + completion_todo_id=completion_todo_id, + autonomous_replan_recorded=effective_autonomous_replan_recorded, + ) + checkpoint_supplement = bool( + refresh_recovery and refresh_recovery["decision"] == "supplement_checkpoint" + ) + if checkpoint_supplement and not vision_checkpoint.get("satisfied"): raise ValueError( - "accountable peer delivery must be refreshed from the independent " - "git worktree that produced it, or name that worktree with " - "--delivery-workspace-path" + "checkpoint supplement did not satisfy the missing decision; " + "an unchanged reason requires an existing vision. Supply a valid vision " + "patch on the same Turn; the original writeback and quota are unchanged." ) - if delivery_workspace_path is not None and delivery_workspace is None: + delivery_workspace = None + workspace_requirement = str( + (settlement_workspace_requirement or {}).get("requirement") or "unknown" + ) + if delivery_workspace_path is not None and workspace_requirement == "not_required": raise ValueError( - "--delivery-workspace-path must identify the registered local goal " - "workspace or a git checkout with a credential-free origin repository" + "--delivery-workspace-path conflicts with the original Todo's " + "explicit non-delivery settlement contract" ) - if checkpoint_supplement: - # The supplemental row must not reattribute the original delivery to - # the recovery caller's current directory or change its accounting. - assert prior_writeback_run is not None - delivery_workspace = prior_writeback_run.get("delivery_workspace") - normalized_delivery_outcome = prior_writeback_run.get("delivery_outcome") - normalized_delivery_batch_scale = prior_writeback_run.get( - "delivery_batch_scale" - ) - normalized_progress_observation = prior_writeback_run.get( - "progress_observation" - ) - classification = prior_writeback_run["classification"] - if ( - active_state_next_action_update - and active_state_next_action_update.get("would_update") - and not dry_run - ): - with exclusive_cross_runtime_file_lock(resolved_state_file): - current_state_text = resolved_state_file.read_text(encoding="utf-8") - if current_state_text != expected_write_state_text: + if ( + turn_scoped_settlement_qualified + and workspace_requirement != "not_required" + and not checkpoint_supplement + ): + delivery_workspace = capture_delivery_workspace( + current_path=delivery_workspace_path, + peer_independent_worktree_required=peer_independent_worktree_required, + local_goal_id=safe_goal_id, + local_project_root=resolved_project, + repository_source=( + "refresh_state.delivery_workspace_path" + if delivery_workspace_path is not None + else None + ), + ) + if ( + peer_independent_worktree_required + and ( + delivery_workspace is None + or delivery_workspace.get("workspace_kind") + != "independent_git_worktree" + ) + ): raise ValueError( - "active goal state changed while refresh-state was qualifying " - "its semantic writeback; retry from the current state" + "accountable peer delivery must be refreshed from the independent " + "git worktree that produced it, or name that worktree with " + "--delivery-workspace-path" ) - require_prose_state_write_allowed( - registry_path=registry_path, runtime_root=runtime_root, - goal_id=safe_goal_id, state_path=resolved_state_file, - original_text=current_state_text, planned_text=state_text, + if delivery_workspace_path is not None and delivery_workspace is None: + raise ValueError( + "--delivery-workspace-path must identify the registered local goal " + "workspace or a git checkout with a credential-free origin repository" + ) + if checkpoint_supplement: + # The supplemental row must not reattribute the original delivery to + # the recovery caller's current directory or change its accounting. + assert prior_writeback_run is not None + delivery_workspace = prior_writeback_run.get("delivery_workspace") + normalized_delivery_outcome = prior_writeback_run.get("delivery_outcome") + normalized_delivery_batch_scale = prior_writeback_run.get( + "delivery_batch_scale" ) - atomic_write_state_text(resolved_state_file, state_text) - record = build_state_refresh_record( - goal_id=safe_goal_id, - state_file=resolved_state_file, - state_text=state_text, - classification=classification, - recommended_action=action, - recommended_action_source=recommended_action_source, - recommended_action_resolution=recommendation_resolution, - generated_at=generated_at, - registry_goal=registry_goal, - delivery_batch_scale=normalized_delivery_batch_scale, - delivery_outcome=normalized_delivery_outcome, - progress_scope=normalized_progress_scope, - agent_id=normalized_agent_id or None, - agent_lane=normalized_agent_lane or None, - autonomous_replan_recorded=effective_autonomous_replan_recorded, - repair_delta_contract=repair_delta_contract, - autonomous_replan_frontier_identity=autonomous_replan_frontier_identity, - agent_vision=agent_vision, - vision_checkpoint=vision_checkpoint, - progress_observation=normalized_progress_observation, - delivery_workspace=delivery_workspace, - settlement_identity=settlement_identity, - ) - if delivery_workspace_causality: - record["delivery_workspace_causality"] = delivery_workspace_causality - if refresh_recovery: - record["refresh_recovery"] = refresh_recovery - if settlement_workspace_requirement: - record["settlement_workspace_requirement"] = ( - settlement_workspace_requirement - ) - if autonomous_replan_recorded: - if "autonomous_replan_ack" not in record: - record["autonomous_replan_ack"] = { - "schema_version": "autonomous_replan_ack_v0", - "recorded": False, - "source": "refresh_state", - "delta_contract": repair_delta_contract, - } - record["autonomous_replan_ack"]["requested"] = True - if autonomous_replan_frontier_identity: - record["autonomous_replan_ack"]["frontier_identity"] = ( - autonomous_replan_frontier_identity + normalized_progress_observation = prior_writeback_run.get( + "progress_observation" ) - if requested_classification != classification: - record["autonomous_replan_ack"]["requested_classification"] = requested_classification - record["autonomous_replan_noop"] = { - "schema_version": REPAIR_NOOP_SCHEMA_VERSION, - "classification": classification, - "requested_classification": requested_classification, - "reason": "autonomous replan ACK requested without a machine-visible repair delta", - } - if replan_semantic_delta: - record.setdefault( - "autonomous_replan_ack", - { - "schema_version": "autonomous_replan_ack_v0", - "recorded": True, - "source": "refresh_state_semantic_delta", - }, - ) - record["autonomous_replan_ack"]["semantic_delta"] = ( - replan_semantic_delta - ) - if active_state_next_action_update: - record["active_state_next_action_update"] = active_state_next_action_update - compact_route = compact_runtime_projection_route(runtime_projection_route) - compact_route["projection_enabled"] = bool(sync_global) - compact_route["projection_marker_field"] = "shared_runtime_projection" - record["runtime_projection_route"] = compact_route - - runs_dir = runtime_root / "goals" / safe_goal_id / "runs" - json_path, markdown_path = unique_run_paths(runs_dir, generated_at) - index_path = runs_dir / "index.jsonl" - index_record, payload = _build_state_refresh_output_projections( - record=record, - registry_path=registry_path, - runtime_root=runtime_root, - project=resolved_project, - json_path=json_path, - markdown_path=markdown_path, - index_path=index_path, - dry_run=dry_run, - autonomous_replan_recorded_requested=bool(autonomous_replan_recorded), - ) - # GH-C95 producer boundary: attach the typed run_usage_v0 row before the - # durable record and index rows are written, so malformed or negative usage - # fails the whole refresh instead of entering run history. The booking lock - # spans ledger-basis read + row append so concurrent refreshes cannot fund - # two deltas from one stale basis; the appended row advances the basis. - with ExitStack() as usage_booking_guard: - if usage_codex_session is not None: - if not dry_run: - runs_dir.mkdir(parents=True, exist_ok=True) - usage_booking_guard.enter_context( - exclusive_file_lock( - usage_booking_lock_target(runs_dir), - agent_id=normalized_agent_id or None, - operation="refresh-state-usage-booking", + classification = prior_writeback_run["classification"] + if ( + active_state_next_action_update + and active_state_next_action_update.get("would_update") + and not dry_run + ): + with exclusive_cross_runtime_file_lock(resolved_state_file): + current_state_text = resolved_state_file.read_text(encoding="utf-8") + if current_state_text != expected_write_state_text: + raise ValueError( + "active goal state changed while refresh-state was qualifying " + "its semantic writeback; retry from the current state" ) + require_prose_state_write_allowed( + registry_path=registry_path, runtime_root=runtime_root, + goal_id=safe_goal_id, state_path=resolved_state_file, + original_text=current_state_text, planned_text=state_text, ) - book_codex_session_usage( - record, usage_codex_session, index_path, index_record=index_record - ) - elif usage_measurement is not None: - ingest_usage_into_run_record( - record, usage_measurement, index_record=index_record + atomic_write_state_text(resolved_state_file, state_text) + record = build_state_refresh_record( + goal_id=safe_goal_id, + state_file=resolved_state_file, + state_text=state_text, + classification=classification, + recommended_action=action, + recommended_action_source=recommended_action_source, + recommended_action_resolution=recommendation_resolution, + generated_at=generated_at, + registry_goal=registry_goal, + delivery_batch_scale=normalized_delivery_batch_scale, + delivery_outcome=normalized_delivery_outcome, + progress_scope=normalized_progress_scope, + agent_id=normalized_agent_id or None, + agent_lane=normalized_agent_lane or None, + autonomous_replan_recorded=effective_autonomous_replan_recorded, + repair_delta_contract=repair_delta_contract, + autonomous_replan_frontier_identity=autonomous_replan_frontier_identity, + agent_vision=agent_vision, + vision_checkpoint=vision_checkpoint, + progress_observation=normalized_progress_observation, + delivery_workspace=delivery_workspace, + settlement_identity=settlement_identity, + ) + if delivery_workspace_causality: + record["delivery_workspace_causality"] = delivery_workspace_causality + if refresh_recovery: + record["refresh_recovery"] = refresh_recovery + if settlement_workspace_requirement: + record["settlement_workspace_requirement"] = ( + settlement_workspace_requirement ) - if isinstance(record.get("usage"), dict): - payload["usage"] = dict(record["usage"]) - if dry_run: - expected_write_scopes = ["runtime_history"] - if active_state_next_action_update and active_state_next_action_update.get("would_update"): - expected_write_scopes.insert(0, "active_state") - if sync_global and route_status in {"resolved", "single_runtime"}: - expected_write_scopes.append("global_registry") - if shared_runtime_root: - expected_write_scopes.append("shared_runtime_projection") - patch_parts = [f"append refresh-state run classification={classification}"] - if active_state_next_action_update: - if active_state_next_action_update.get("would_update"): - patch_parts.append("preview active-state Next Action update") - else: - patch_parts.append("preserve active-state Next Action") - if sync_global and route_status in {"resolved", "single_runtime"}: - patch_parts.append("sync public-safe registry projection") - elif sync_global: - patch_parts.append(f"block global sync on {route_status} runtime projection route") - if shared_runtime_root: - patch_parts.append("project compact refresh to registered shared runtime") - payload["local_state_write_correctness"] = build_local_state_write_correctness_dry_run_packet( - goal_id=safe_goal_id, - writer_id=normalized_agent_id or "loopx.refresh-state", - write_class="refresh_state", - state_text=expected_write_state_text, - target_refs={ - "state_file_ref": "registry.goal.state_file", - "run_history_ref": "runtime.goal.runs", - "index_ref": "runtime.goal.runs.index", - "global_registry_ref": ( - "runtime.registry.global" - if sync_global and route_status in {"resolved", "single_runtime"} - else None - ), - "shared_runtime_projection_ref": ( - "shared_runtime.goal.runs.index" if shared_runtime_root else None - ), + if autonomous_replan_recorded: + if "autonomous_replan_ack" not in record: + record["autonomous_replan_ack"] = { + "schema_version": "autonomous_replan_ack_v0", + "recorded": False, + "source": "refresh_state", + "delta_contract": repair_delta_contract, + } + record["autonomous_replan_ack"]["requested"] = True + if autonomous_replan_frontier_identity: + record["autonomous_replan_ack"]["frontier_identity"] = ( + autonomous_replan_frontier_identity + ) + if requested_classification != classification: + record["autonomous_replan_ack"]["requested_classification"] = requested_classification + record["autonomous_replan_noop"] = { + "schema_version": REPAIR_NOOP_SCHEMA_VERSION, + "classification": classification, + "requested_classification": requested_classification, + "reason": "autonomous replan ACK requested without a machine-visible repair delta", + } + if replan_semantic_delta: + record.setdefault( + "autonomous_replan_ack", + { + "schema_version": "autonomous_replan_ack_v0", + "recorded": True, + "source": "refresh_state_semantic_delta", }, - patch_summary="; ".join(patch_parts), - expected_write_scopes=expected_write_scopes, - lease_ref=None, - projection_status_surface=f"refresh-state dry-run: {classification}", ) - if not dry_run: - runs_dir.mkdir(parents=True, exist_ok=True) - json_path, markdown_path = reserve_unique_run_paths(runs_dir, generated_at) - index_record["json_path"] = str(json_path) - index_record["markdown_path"] = str(markdown_path) - payload["json_path"] = str(json_path) - payload["markdown_path"] = str(markdown_path) - json_path.write_text( - json.dumps(record, ensure_ascii=False, indent=2, allow_nan=False) + "\n", - encoding="utf-8", + record["autonomous_replan_ack"]["semantic_delta"] = ( + replan_semantic_delta ) - markdown_path.write_text(render_state_refresh_markdown(payload) + "\n", encoding="utf-8") - with index_path.open("a", encoding="utf-8") as f: - f.write(json.dumps(index_record, ensure_ascii=False, allow_nan=False) + "\n") - if sync_global and route_status in {"missing", "ambiguous"}: - payload["ok"] = False - payload["partial_write"] = not dry_run - payload["global_sync"] = { - "ok": False, - "enabled": False, - "wrote": False, - "reason": f"runtime projection route is {route_status}", - "route_status": route_status, - } - payload["shared_runtime_projection"] = { - "ok": False, - "status": f"route_{route_status}", - "dry_run": dry_run, - "raw_artifacts_copied": False, - "recommended_action_copied": False, - "runtime_projection_route_id": compact_route.get("route_id"), - } - elif sync_global: - payload["global_sync"] = sync_project_registry_to_global( + if active_state_next_action_update: + record["active_state_next_action_update"] = active_state_next_action_update + compact_route = compact_runtime_projection_route(runtime_projection_route) + compact_route["projection_enabled"] = bool(sync_global) + compact_route["projection_marker_field"] = "shared_runtime_projection" + record["runtime_projection_route"] = compact_route + + runs_dir = runtime_root / "goals" / safe_goal_id / "runs" + json_path, markdown_path = unique_run_paths(runs_dir, generated_at) + index_path = runs_dir / "index.jsonl" + index_record, payload = _build_state_refresh_output_projections( + record=record, registry_path=registry_path, - runtime_root_override=str(global_sync_runtime_root or runtime_root), - goal_id=safe_goal_id, + runtime_root=runtime_root, + project=resolved_project, + json_path=json_path, + markdown_path=markdown_path, + index_path=index_path, dry_run=dry_run, + autonomous_replan_recorded_requested=bool(autonomous_replan_recorded), ) - if shared_runtime_root and payload["global_sync"].get("ok"): - projection_record, projection_index = build_shared_runtime_projection( - record=record, - ) - try: - payload["shared_runtime_projection"] = write_shared_runtime_projection( - shared_runtime_root=shared_runtime_root, + # GH-C95 producer boundary: attach the typed run_usage_v0 row before the + # durable record and index rows are written, so malformed or negative usage + # fails the whole refresh instead of entering run history. The booking lock + # spans ledger-basis read + row append so concurrent refreshes cannot fund + # two deltas from one stale basis; the appended row advances the basis. + with ExitStack() as usage_booking_guard: + if usage_codex_session is not None: + if not dry_run: + runs_dir.mkdir(parents=True, exist_ok=True) + usage_booking_guard.enter_context( + exclusive_file_lock( + usage_booking_lock_target(runs_dir), + agent_id=normalized_agent_id or None, + operation="refresh-state-usage-booking", + ) + ) + book_codex_session_usage( + record, usage_codex_session, index_path, index_record=index_record + ) + elif usage_measurement is not None: + ingest_usage_into_run_record( + record, usage_measurement, index_record=index_record + ) + if isinstance(record.get("usage"), dict): + payload["usage"] = dict(record["usage"]) + if dry_run: + expected_write_scopes = ["runtime_history"] + if active_state_next_action_update and active_state_next_action_update.get("would_update"): + expected_write_scopes.insert(0, "active_state") + if sync_global and route_status in {"resolved", "single_runtime"}: + expected_write_scopes.append("global_registry") + if shared_runtime_root: + expected_write_scopes.append("shared_runtime_projection") + patch_parts = [f"append refresh-state run classification={classification}"] + if active_state_next_action_update: + if active_state_next_action_update.get("would_update"): + patch_parts.append("preview active-state Next Action update") + else: + patch_parts.append("preserve active-state Next Action") + if sync_global and route_status in {"resolved", "single_runtime"}: + patch_parts.append("sync public-safe registry projection") + elif sync_global: + patch_parts.append(f"block global sync on {route_status} runtime projection route") + if shared_runtime_root: + patch_parts.append("project compact refresh to registered shared runtime") + payload["local_state_write_correctness"] = build_local_state_write_correctness_dry_run_packet( goal_id=safe_goal_id, - record=projection_record, - index_record=projection_index, - dry_run=dry_run, + writer_id=normalized_agent_id or "loopx.refresh-state", + write_class="refresh_state", + state_text=expected_write_state_text, + target_refs={ + "state_file_ref": "registry.goal.state_file", + "run_history_ref": "runtime.goal.runs", + "index_ref": "runtime.goal.runs.index", + "global_registry_ref": ( + "runtime.registry.global" + if sync_global and route_status in {"resolved", "single_runtime"} + else None + ), + "shared_runtime_projection_ref": ( + "shared_runtime.goal.runs.index" if shared_runtime_root else None + ), + }, + patch_summary="; ".join(patch_parts), + expected_write_scopes=expected_write_scopes, + lease_ref=None, + projection_status_surface=f"refresh-state dry-run: {classification}", + ) + if not dry_run: + runs_dir.mkdir(parents=True, exist_ok=True) + json_path, markdown_path = reserve_unique_run_paths(runs_dir, generated_at) + index_record["json_path"] = str(json_path) + index_record["markdown_path"] = str(markdown_path) + payload["json_path"] = str(json_path) + payload["markdown_path"] = str(markdown_path) + json_path.write_text( + json.dumps(record, ensure_ascii=False, indent=2, allow_nan=False) + "\n", + encoding="utf-8", + ) + markdown_path.write_text(render_state_refresh_markdown(payload) + "\n", encoding="utf-8") + with index_path.open("a", encoding="utf-8") as f: + f.write(json.dumps(index_record, ensure_ascii=False, allow_nan=False) + "\n") + if sync_global and route_status in {"missing", "ambiguous"}: + payload["ok"] = False + payload["partial_write"] = not dry_run + payload["global_sync"] = { + "ok": False, + "enabled": False, + "wrote": False, + "reason": f"runtime projection route is {route_status}", + "route_status": route_status, + } + payload["shared_runtime_projection"] = { + "ok": False, + "status": f"route_{route_status}", + "dry_run": dry_run, + "raw_artifacts_copied": False, + "recommended_action_copied": False, + "runtime_projection_route_id": compact_route.get("route_id"), + } + elif sync_global: + payload["global_sync"] = sync_project_registry_to_global( + registry_path=registry_path, + runtime_root_override=str(global_sync_runtime_root or runtime_root), + goal_id=safe_goal_id, + dry_run=dry_run, + ) + if shared_runtime_root and payload["global_sync"].get("ok"): + projection_record, projection_index = build_shared_runtime_projection( + record=record, ) - except OSError as exc: + try: + payload["shared_runtime_projection"] = write_shared_runtime_projection( + shared_runtime_root=shared_runtime_root, + goal_id=safe_goal_id, + record=projection_record, + index_record=projection_index, + dry_run=dry_run, + ) + except OSError as exc: + payload["ok"] = False + payload["partial_write"] = not dry_run + payload["shared_runtime_projection"] = { + "ok": False, + "status": "write_failed", + "dry_run": dry_run, + "shared_runtime_root": str(shared_runtime_root), + "raw_artifacts_copied": False, + "recommended_action_copied": False, + "error": str(exc), + } + elif shared_runtime_root: payload["ok"] = False payload["partial_write"] = not dry_run payload["shared_runtime_projection"] = { "ok": False, - "status": "write_failed", + "status": "blocked_by_global_sync", "dry_run": dry_run, "shared_runtime_root": str(shared_runtime_root), "raw_artifacts_copied": False, "recommended_action_copied": False, - "error": str(exc), } - elif shared_runtime_root: - payload["ok"] = False - payload["partial_write"] = not dry_run - payload["shared_runtime_projection"] = { - "ok": False, - "status": "blocked_by_global_sync", - "dry_run": dry_run, - "shared_runtime_root": str(shared_runtime_root), - "raw_artifacts_copied": False, - "recommended_action_copied": False, - } + else: + payload["shared_runtime_projection"] = { + "ok": True, + "status": "not_required", + "dry_run": dry_run, + "raw_artifacts_copied": False, + "recommended_action_copied": False, + } else: + payload["global_sync"] = { + "enabled": False, + "global_registry": str(runtime_root / "registry.global.json"), + "synced_goal_ids": [], + "wrote": False, + } payload["shared_runtime_projection"] = { "ok": True, - "status": "not_required", + "status": "disabled", "dry_run": dry_run, "raw_artifacts_copied": False, "recommended_action_copied": False, } - else: - payload["global_sync"] = { - "enabled": False, - "global_registry": str(runtime_root / "registry.global.json"), - "synced_goal_ids": [], - "wrote": False, - } - payload["shared_runtime_projection"] = { - "ok": True, - "status": "disabled", - "dry_run": dry_run, - "raw_artifacts_copied": False, - "recommended_action_copied": False, - } - return payload + return payload diff --git a/tests/control_plane/test_delivery_claim_validation.py b/tests/control_plane/test_delivery_claim_validation.py index d100fb0d7a..d6086eecf0 100644 --- a/tests/control_plane/test_delivery_claim_validation.py +++ b/tests/control_plane/test_delivery_claim_validation.py @@ -29,6 +29,8 @@ def test_historical_conflict_is_visible_and_new_write_has_no_artifacts(tmp_path, def test_refresh_rejects_primary_blocker_before_loading_registry_or_mutating_state(tmp_path): with pytest.raises(ValueError, match="primary_outcome_with_blocker"): refresh_state_run(registry_path=tmp_path / "absent-registry.json", goal_id="delivery", + runtime_root_override=None, project=None, state_file=None, recommended_action=None, + dry_run=False, classification="synthetic_delivery", delivery_outcome="primary_goal_outcome", todo_id="todo_delivery", progress_observation={ "schema_version": "typed_progress_observation_v0", "result_class": "blocked", @@ -37,6 +39,29 @@ def test_refresh_rejects_primary_blocker_before_loading_registry_or_mutating_sta assert list(tmp_path.iterdir()) == [] +@pytest.mark.parametrize("dry_run", [False, True]) +@pytest.mark.parametrize("overrides, error", [ + ({"delivery_outcome": "unsupported_outcome", "progress_observation": {"schema_version": "bad"}}, + "delivery_outcome must be one of"), + ({"delivery_batch_scale": "unsupported_scale", "delivery_outcome": "unsupported_outcome"}, + "delivery_batch_scale must be one of"), + ({"agent_lane": "lane-a", "delivery_outcome": "unsupported_outcome"}, + "--agent-lane requires --agent-id"), + ({"delivery_outcome": "primary_goal_outcome", "progress_observation": {"schema_version": "bad"}}, + "progress observation must use"), +]) +def test_refresh_validates_fields_in_order_before_cross_field_rules_or_io(tmp_path, dry_run, overrides, error): + # An absent registry also proves that malformed input is rejected before + # opening the store or creating its write lock, in both execution modes. + with pytest.raises(ValueError, match=error): + refresh_state_run( + registry_path=tmp_path / "absent-registry.json", runtime_root_override=None, + goal_id="delivery", project=None, state_file=None, recommended_action=None, + classification="synthetic_delivery", dry_run=dry_run, **overrides, + ) + assert list(tmp_path.iterdir()) == [] + + def test_valid_partial_progress_and_unspecified_delivery_remain_legal(): for record in ({}, {"delivery_outcome": "outcome_progress", "delivery_turn_kind": "product_path_execution"}, {"delivery_outcome": "outcome_gap", "delivery_turn_kind": "blocker_writeback"}): From f66310806321b4c762565c144ba446ebe9a9350c Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 9 Sep 2026 17:08:46 +0800 Subject: [PATCH 3/3] test(refresh): keep CAS mutation independent of indentation Signed-off-by: huangruiteng --- examples/shared-goal-authority-e2e/mutants.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/examples/shared-goal-authority-e2e/mutants.py b/examples/shared-goal-authority-e2e/mutants.py index aab87183e2..3ad623db37 100644 --- a/examples/shared-goal-authority-e2e/mutants.py +++ b/examples/shared-goal-authority-e2e/mutants.py @@ -203,11 +203,8 @@ def apply(source: str) -> str: move_guard_outside_lock("require_registry_source_write_allowed")),), WRITER_TEST + "test_waiting_override_writer_rechecks_registry_binding_inside_shared_state_lock"), Case("remove_refresh_cas", (("loopx/state_refresh.py", replacement( - ''' if current_state_text != expected_write_state_text: - raise ValueError( - "active goal state changed while refresh-state was qualifying " - "its semantic writeback; retry from the current state" - )''', "")),), + "if current_state_text != expected_write_state_text:", + "if False: # DELIBERATE MUTANT: bypass stale-state rejection.")),), WRITER_TEST + "test_concurrent_public_refresh_preserves_the_newer_owned_paragraph"), Case("fence_unshared_state_lock", ((COORDINATION + "legacy_writer_fence.ts", replacement( "withFileMutationLock(statePath, () =>",