From ad47c8e54f0c4e624db17244e6eb3814f2377562 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Tue, 22 Sep 2026 17:50:49 +0800 Subject: [PATCH 1/3] refactor(coordination): unify handoff transition facts and recovery Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../coordination/handoff_mode_facts.ts | 51 ++++ .../coordination/handoff_mode_legacy_plan.ts | 74 +++++ .../coordination/handoff_mode_policy.ts | 17 +- .../coordination/handoff_mode_transaction.ts | 68 ++--- .../control_plane/effect_runtime_handlers.ts | 2 + loopx/control_plane/todos/handoff_mode.py | 281 ++++-------------- .../todos/handoff_mode_source.py | 61 ++++ 7 files changed, 285 insertions(+), 269 deletions(-) create mode 100644 loopx/control_plane/coordination/handoff_mode_facts.ts create mode 100644 loopx/control_plane/coordination/handoff_mode_legacy_plan.ts create mode 100644 loopx/control_plane/todos/handoff_mode_source.py diff --git a/loopx/control_plane/coordination/handoff_mode_facts.ts b/loopx/control_plane/coordination/handoff_mode_facts.ts new file mode 100644 index 0000000000..cdad54d7c9 --- /dev/null +++ b/loopx/control_plane/coordination/handoff_mode_facts.ts @@ -0,0 +1,51 @@ +/** Quiescence is a decision over complete ownership facts, never a display page. */ +import type {JsonObject} from "../effect_program.ts"; +import {requireJsonObject} from "../runtime_decode.ts"; +import {parseIsoTimestamp} from "../runtime_timestamp.ts"; +import {leaseIsActive, TASK_LEASE_SCHEMA_VERSION} from "../work_items/task_lease_acquire.ts"; +import {HANDOFF_MODES, type HandoffMode} from "./handoff_mode_policy.ts"; + +export type PersistedHandoffMode = + | {kind: "valid"; value: HandoffMode} + | {kind: "invalid"; value: string}; + +export function persistedHandoffMode(value: unknown): PersistedHandoffMode { + const text = value == null ? "" : String(value).trim(); + if (!text) return {kind: "valid", value: "legacy"}; + const mode = HANDOFF_MODES.find(mode => mode === text); + return mode ? {kind: "valid", value: mode} : {kind: "invalid", value: text}; +} + +export function previousModeFields(mode: PersistedHandoffMode): JsonObject { + return {previous_mode: mode.value, previous_mode_valid: mode.kind === "valid", + ...(mode.kind === "invalid" ? {previous_mode_error_code: "invalid_handoff_mode"} : {})}; +} + +export interface HandoffQuiescence { + claimed_todos: JsonObject[]; + active_leases: JsonObject[]; +} + +export function handoffQuiescence(todos: readonly JsonObject[], leases: readonly JsonObject[], + observedAt: string): HandoffQuiescence { + const now = parseIsoTimestamp(observedAt); + if (!now) throw new Error("observed_at must be a valid ISO timestamp"); + const claimed: JsonObject[] = []; + for (const value of todos) { + const todo = requireJsonObject(value, "handoff Todo"); + if (todo.archive_state === "archive" || todo.done === true) continue; + if (todo.claimed_by == null || todo.claimed_by === "") continue; + if (typeof todo.claimed_by !== "string") throw new Error("Todo claimed_by must be a string or null"); + const owner = todo.claimed_by.trim(); + if (owner) claimed.push({todo_id: todo.todo_id ?? null, claimed_by: owner, status: todo.status ?? null}); + } + const active: JsonObject[] = []; + for (const value of leases) { + const lease = requireJsonObject(value, "handoff lease"); + if (lease.schema_version !== TASK_LEASE_SCHEMA_VERSION) throw new Error("lease schema mismatch"); + if (leaseIsActive(lease, now)) active.push({todo_id: lease.todo_id ?? null, + owner: lease.owner ?? null, expires_at: lease.expires_at ?? null, + ...(typeof lease.lease_path === "string" ? {lease_path: lease.lease_path} : {})}); + } + return {claimed_todos: claimed, active_leases: active}; +} diff --git a/loopx/control_plane/coordination/handoff_mode_legacy_plan.ts b/loopx/control_plane/coordination/handoff_mode_legacy_plan.ts new file mode 100644 index 0000000000..a1e9e40e37 --- /dev/null +++ b/loopx/control_plane/coordination/handoff_mode_legacy_plan.ts @@ -0,0 +1,74 @@ +/** Typed compatibility plan. Python owns source locks/capture; TS owns the change. */ +import type {JsonObject} from "../effect_program.ts"; +import {requireJsonObject, requireStringLiteral} from "../runtime_decode.ts"; +import {HANDOFF_MODES, decideHandoffMode} from "./handoff_mode_policy.ts"; +import {handoffQuiescence, persistedHandoffMode, previousModeFields} from "./handoff_mode_facts.ts"; + +export const LEGACY_HANDOFF_PLAN_SCHEMA = "loopx_legacy_handoff_mode_plan_request_v0"; +const RESULT_SCHEMA = "loopx_legacy_handoff_mode_plan_result_v0"; + +/** Edit only the generated scalar field. All other bytes, including CRLF and + * Unicode line separators inside quoted metadata, remain untouched. A duplicate + * key has no single writable owner, so do not silently patch only one copy. */ +export function patchHandoffMode(text: string, requested: string): string { + const opening = /^---[ \t]*\r?\n/.exec(text)?.[0]; + if (!opening) throw new Error("state_frontmatter_missing"); + const fields: {start: number; end: number; ending: string}[] = []; + let close: number | undefined; + // JavaScript multiline ^ also recognizes U+2028/U+2029. The generated + // metadata protocol uses physical LF only, including inside JSON strings. + for (let start = opening.length; start <= text.length;) { + const newline = text.indexOf("\n", start); + const end = newline < 0 ? text.length : newline + 1; + const raw = text.slice(start, end); + const line = raw.replace(/\r?\n$/, ""); + if (/^---[ \t]*$/.test(line)) { close = start; break; } + if (/^[ \t]*handoff_mode[ \t]*:/.test(line)) { + fields.push({start, end, ending: /\r?\n$/.exec(raw)?.[0] ?? ""}); + } + if (newline < 0) break; + start = end; + } + if (close === undefined) throw new Error("state_frontmatter_missing"); + if (fields.length > 1) throw new Error("handoff_mode_duplicate_field"); + const field = fields[0]; + if (field) return text.slice(0, field.start) + `handoff_mode: ${requested}${field.ending}` + text.slice(field.end); + const newline = opening.endsWith("\r\n") ? "\r\n" : "\n"; + return text.slice(0, close) + `handoff_mode: ${requested}${newline}` + text.slice(close); +} + +export function planLegacyHandoffMode(value: unknown): JsonObject { + const input = requireJsonObject(value, "legacy handoff mode plan"); + if (input.schema_version !== LEGACY_HANDOFF_PLAN_SCHEMA) throw new Error("legacy handoff plan schema mismatch"); + const requested = requireStringLiteral(input.requested_mode, HANDOFF_MODES, "requested_mode"); + if (typeof input.frontmatter_text !== "string") throw new Error("frontmatter_text must be a string"); + const previous = persistedHandoffMode(input.previous_value); + const fields = {...previousModeFields(previous), handoff_mode: requested}; + // An identical valid mode grants no new behavior. Preserve the compatibility + // no-op even when leases exist or the legacy state has no frontmatter. + if (previous.kind === "valid" && previous.value === requested) { + return {schema_version: RESULT_SCHEMA, outcome: "no_change", code: "handoff_mode_unchanged", + changed: false, ...fields}; + } + if (input.todos === null && input.leases === null) return {schema_version: RESULT_SCHEMA, + outcome: "snapshot_required", ...fields, changed: false}; + if (!Array.isArray(input.todos) || !Array.isArray(input.leases)) throw new Error("complete Todo and lease facts required"); + let facts; + try { + facts = handoffQuiescence(input.todos.map(row => requireJsonObject(row, "Todo")), + input.leases.map(row => requireJsonObject(row, "lease")), String(input.observed_at)); + } catch (error) { + return {schema_version: RESULT_SCHEMA, outcome: "rejected", code: "invalid_handoff_mode_authority", + ...fields, changed: false, reason: String(error)}; + } + const plan = decideHandoffMode(previous, requested, facts.claimed_todos.length, facts.active_leases.length); + if (plan.outcome === "rejected") return {schema_version: RESULT_SCHEMA, ...plan, ...fields, + changed: false, ...facts}; + try { + return {schema_version: RESULT_SCHEMA, ...plan, ...fields, changed: true, + next_frontmatter_text: patchHandoffMode(input.frontmatter_text, requested)}; + } catch (error) { + if (!(error instanceof Error) || !["state_frontmatter_missing", "handoff_mode_duplicate_field"].includes(error.message)) throw error; + return {schema_version: RESULT_SCHEMA, outcome: "rejected", code: error.message, ...fields, changed: false}; + } +} diff --git a/loopx/control_plane/coordination/handoff_mode_policy.ts b/loopx/control_plane/coordination/handoff_mode_policy.ts index dd978bd61b..0ba00bd379 100644 --- a/loopx/control_plane/coordination/handoff_mode_policy.ts +++ b/loopx/control_plane/coordination/handoff_mode_policy.ts @@ -6,6 +6,17 @@ export const HANDOFF_MODES = ["legacy", "soft_claim", "hard_lease"] as const; export type HandoffMode = typeof HANDOFF_MODES[number]; export const HANDOFF_MODE_PLAN_SCHEMA = "loopx_handoff_mode_plan_request_v0"; +import type {PersistedHandoffMode} from "./handoff_mode_facts.ts"; + +export function decideHandoffMode(previous: PersistedHandoffMode, requested: HandoffMode, + claims: number, leases: number): JsonObject { + const unchanged = previous.kind === "valid" && previous.value === requested; + const rejected = !unchanged && (claims > 0 || leases > 0); + return {outcome: unchanged ? "no_change" : rejected ? "rejected" : "apply", + code: unchanged ? "handoff_mode_unchanged" : rejected ? "handoff_mode_not_quiescent" : "handoff_mode_transition", + idempotent: unchanged, previous_mode: previous.value, handoff_mode: requested}; +} + export function planHandoffMode(value: unknown): JsonObject { const input = requireJsonObject(value, "handoff mode plan"); if (input.schema_version !== HANDOFF_MODE_PLAN_SCHEMA) throw new Error("handoff mode plan schema mismatch"); @@ -13,10 +24,6 @@ export function planHandoffMode(value: unknown): JsonObject { const requested = requireStringLiteral(input.requested_mode, HANDOFF_MODES, "requested_mode"); const claims = requireStringArray(input.active_claimed_todo_ids, "active_claimed_todo_ids"); const leases = requireStringArray(input.active_lease_todo_ids, "active_lease_todo_ids"); - const unchanged = previous === requested; - const rejected = !unchanged && (claims.length > 0 || leases.length > 0); return {schema_version: "loopx_handoff_mode_plan_result_v0", - outcome: unchanged ? "no_change" : rejected ? "rejected" : "apply", - code: unchanged ? "handoff_mode_unchanged" : rejected ? "handoff_mode_not_quiescent" : "handoff_mode_transition", - idempotent: unchanged, previous_mode: previous, handoff_mode: requested}; + ...decideHandoffMode({kind: "valid", value: previous}, requested, claims.length, leases.length)}; } diff --git a/loopx/control_plane/coordination/handoff_mode_transaction.ts b/loopx/control_plane/coordination/handoff_mode_transaction.ts index d5b7feb15f..5009768827 100644 --- a/loopx/control_plane/coordination/handoff_mode_transaction.ts +++ b/loopx/control_plane/coordination/handoff_mode_transaction.ts @@ -1,12 +1,13 @@ /** Provider-neutral mode transition: quiescence and mode share one CAS snapshot. */ import type {JsonObject} from "../effect_program.ts"; -import type {AuthorityStore, AuthorityStoreReceiptResult} from "./authority_store.ts"; -import {canonicalAuthorityObject, canonicalAuthoritySha256, requireAuthorityStoreId} from "./authority_store_codec.ts"; +import type {AuthorityStore} from "./authority_store.ts"; +import {AuthorityStoreProtocolError, canonicalAuthorityObject, canonicalAuthoritySha256, requireAuthorityStoreId} from "./authority_store_codec.ts"; import {indexCoordinationProjection, validateCoordinationTodoReadModel} from "./coordination_projection.ts"; -import {HANDOFF_MODES, HANDOFF_MODE_PLAN_SCHEMA, planHandoffMode} from "./handoff_mode_policy.ts"; +import {HANDOFF_MODES, decideHandoffMode} from "./handoff_mode_policy.ts"; import {requireBoolean, requireStringLiteral} from "../runtime_decode.ts"; import {parseIsoTimestamp} from "../runtime_timestamp.ts"; -import {leaseIsActive, TASK_LEASE_SCHEMA_VERSION} from "../work_items/task_lease_acquire.ts"; +import {handoffQuiescence} from "./handoff_mode_facts.ts"; +import {CoordinationCommandReceipt} from "./command_receipt.ts"; export const HANDOFF_MODE_SET_SCHEMA = "loopx_coordination_handoff_mode_set_request_v0"; const RESULT_SCHEMA = "loopx_coordination_handoff_mode_set_result_v0"; @@ -19,30 +20,30 @@ export interface HandoffModeSetInput { dry_run: boolean; } -function failure(reason_code: string, reason: string, failureKind?: "decision_rejection"): JsonObject { +function failure(reason_code: string, reason: string, failureKind?: "decision_rejection"): JsonObject & {schema_version: typeof RESULT_SCHEMA} { return {schema_version: RESULT_SCHEMA, status: "failed", changed: false, reason_code, reason, ...(failureKind ? {failure_kind: failureKind} : {})}; } -function replay(receipt: AuthorityStoreReceiptResult, input: HandoffModeSetInput, hash: string, - status: "applied" | "replayed" | "recovered"): JsonObject | null { - if (receipt.status === "missing") return null; - if (receipt.status !== "found") return {schema_version: RESULT_SCHEMA, ...receipt, changed: false}; - const record = receipt.receipts[0]; - if (receipt.receipts.length !== 1 || record?.schema_version !== RECEIPT_SCHEMA || - record.goal_id !== input.goal_id || record.operation_id !== input.operation_id || record.request_sha256 !== hash) { - return failure("coordination_operation_identity_mismatch", "operation id names another handoff mode intent", - "decision_rejection"); - } - const decision = canonicalAuthorityObject(record.decision, "handoff mode decision receipt"); - return {schema_version: RESULT_SCHEMA, ...decision, status, - changed: status !== "replayed" && decision.changed === true, - provider_revision: receipt.provider_revision, cursor: receipt.cursor}; +function commandReceipt(input: HandoffModeSetInput, hash: string) { + return new CoordinationCommandReceipt({result_schema: RESULT_SCHEMA, + identity: {schema_version: RECEIPT_SCHEMA, goal_id: input.goal_id, + operation_id: input.operation_id, request_sha256: hash}, + failure: (code, reason) => failure(code, reason, "decision_rejection"), + decode(record) { + const decision = canonicalAuthorityObject(record.decision, "handoff mode decision receipt"); + if (typeof decision.changed !== "boolean" || decision.goal_id !== input.goal_id || + decision.operation_id !== input.operation_id || decision.handoff_mode !== input.requested_mode || + decision.previous_mode_valid !== true || + !HANDOFF_MODES.some(mode => mode === decision.previous_mode)) { + throw new AuthorityStoreProtocolError("invalid handoff mode decision receipt"); + } + return {fields: decision, changed: decision.changed}; + }}); } export async function executeHandoffModeSet(store: AuthorityStore, raw: HandoffModeSetInput): Promise { let input: HandoffModeSetInput; - let now: Date; try { input = {...raw, goal_id: requireAuthorityStoreId(raw.goal_id, "goal id"), operation_id: requireAuthorityStoreId(raw.operation_id, "operation id"), @@ -50,12 +51,12 @@ export async function executeHandoffModeSet(store: AuthorityStore, raw: HandoffM dry_run: requireBoolean(raw.dry_run, "dry_run")}; const parsed = typeof input.observed_at === "string" ? parseIsoTimestamp(input.observed_at) : null; if (!parsed) throw new Error("observed_at must be a valid ISO timestamp"); - now = parsed; } catch (error) { return failure("invalid_handoff_mode_request", String(error)); } // Retry time is observation context, not a new intent. Preview never consumes an operation id. const hash = canonicalAuthoritySha256({goal_id: input.goal_id, requested_mode: input.requested_mode}); + const receipt = commandReceipt(input, hash); if (!input.dry_run) { - const previous = replay(await store.readReceipt(input.operation_id), input, hash, "replayed"); + const previous = await receipt.read(store); if (previous) return previous; } const loaded = await store.loadAuthority(); @@ -66,16 +67,10 @@ export async function executeHandoffModeSet(store: AuthorityStore, raw: HandoffM const indexed = indexCoordinationProjection(head, input.goal_id); validateCoordinationTodoReadModel(head, input.goal_id); const previous = requireStringLiteral(head.handoff_mode ?? "legacy", HANDOFF_MODES, "canonical handoff_mode"); - const claimed = [...indexed.todos.values()].filter(todo => todo.archive_state === "active" && - todo.done !== true && typeof todo.claimed_by === "string" && todo.claimed_by.trim()).map(todo => ({ - todo_id: todo.todo_id, claimed_by: todo.claimed_by, status: todo.status})); - const leases = [...indexed.leases.values()].filter(lease => { - if (lease.schema_version !== TASK_LEASE_SCHEMA_VERSION) throw new Error("canonical lease schema mismatch"); - return leaseIsActive(lease, now); - }).map(lease => ({todo_id: lease.todo_id, owner: lease.owner, expires_at: lease.expires_at})); - const plan = planHandoffMode({schema_version: HANDOFF_MODE_PLAN_SCHEMA, previous_mode: previous, - requested_mode: input.requested_mode, active_claimed_todo_ids: claimed.map(todo => todo.todo_id), - active_lease_todo_ids: leases.map(lease => lease.todo_id)}); + const {claimed_todos: claimed, active_leases: leases} = handoffQuiescence( + [...indexed.todos.values()], [...indexed.leases.values()], input.observed_at); + const plan = decideHandoffMode({kind: "valid", value: previous}, + requireStringLiteral(input.requested_mode, HANDOFF_MODES, "requested_mode"), claimed.length, leases.length); decision = {goal_id: input.goal_id, operation_id: input.operation_id, previous_mode: previous, previous_mode_valid: true, handoff_mode: input.requested_mode, changed: plan.outcome === "apply"}; if (plan.outcome === "rejected") return {...failure(String(plan.code), @@ -86,14 +81,13 @@ export async function executeHandoffModeSet(store: AuthorityStore, raw: HandoffM if (input.dry_run) return {schema_version: RESULT_SCHEMA, ...decision, status: "planned", provider_revision: loaded.provider_revision}; // Seal even an unchanged accepted intent: retry after another mode switch must not reapply it. - const commit = await store.commitAuthority({operation_id: input.operation_id, + const result = await receipt.commit(store, {operation_id: input.operation_id, expected_provider_revision: loaded.provider_revision, next_projection: {...loaded.head, handoff_mode: input.requested_mode}, events: decision.changed ? [{schema_version: "loopx_handoff_mode_changed_v0", ...decision}] : [], receipts: [{schema_version: RECEIPT_SCHEMA, goal_id: input.goal_id, operation_id: input.operation_id, request_sha256: hash, decision}]}); - return replay(await store.readReceipt(input.operation_id), input, hash, - commit.status === "applied" ? "applied" : "recovered") ?? (commit.status === "applied" - ? failure("coordination_commit_readback_mismatch", "applied mode transaction lacks its durable receipt") - : {schema_version: RESULT_SCHEMA, ...commit, changed: false}); + // This command historically reports a committed no-op as applied. Preserve + // that wire contract while sharing durable recovery and strict receipt decode. + return result.status === "no_change" ? {...result, status: "applied"} : result; } diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 2fa4ff5c46..ec964568db 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -13,6 +13,7 @@ import {previewTeamPlan, planTeamTransaction, teamTransactionIdentity} from "./w import {commitLocalTeamPlan} from "./work_items/team_plan_authority.ts"; import {inspectLocalGoalAcceptance, commitLocalGoalAcceptance, commitLocalGoalAcceptanceVerification} from "./goals/acceptance_authority.ts"; +import {planLegacyHandoffMode} from "./coordination/handoff_mode_legacy_plan.ts"; import {planHandoffMode} from "./coordination/handoff_mode_policy.ts"; import {setLocalHandoffMode} from "./coordination/handoff_mode_runtime.ts"; import {projectOwnershipObservation} from "./coordination/ownership_observation.ts"; @@ -544,6 +545,7 @@ export function createEffectRuntimeHandlers( ["coordination.local_authority.todo_update", updateLocalCoordinationTodo], ["coordination.local_authority.monitor_poll", pollLocalCoordinationMonitor], ["coordination.handoff_mode.plan", planHandoffMode], + ["coordination.handoff_mode.legacy_plan", planLegacyHandoffMode], ["coordination.local_authority.handoff_mode_set", setLocalHandoffMode], ["coordination.local_authority.todo_terminal", terminalLifecycleLocalCoordinationTodo], ["coordination.local_authority.todo_archive", archiveLocalCoordinationTodos], diff --git a/loopx/control_plane/todos/handoff_mode.py b/loopx/control_plane/todos/handoff_mode.py index 3bf3f9b71a..367c8b2bb5 100644 --- a/loopx/control_plane/todos/handoff_mode.py +++ b/loopx/control_plane/todos/handoff_mode.py @@ -22,13 +22,9 @@ transaction. Stale or missing Markdown and local lease files are not fallback sources. The legacy mode below remains a frontmatter compatibility contract. -The unpromoted v0 transition scan is materialized-state only: it reads open claims from -the locked ``ACTIVE_GOAL_STATE.md`` text plus time-active local lease files. It -does not merge the event projection, so a claim that exists only in the event -log can be missed. A successful switch is therefore not a proof that every -projection is quiescent. The selected mode's typed per-write gate remains the -safety boundary for later governed ownership and completion mutations, -including event-projected completion. +Unpromoted transitions read the complete Todo event overlay under the append +store locks and local leases under the lease mutex. The same typed quiescence +rule governs both paths; malformed event sources cannot become empty evidence. """ from __future__ import annotations @@ -38,15 +34,11 @@ from typing import Any from ..coordination.authority_core import ( - CoordinationSnapshot, - DecisionOutcome, HandoffMode, - HandoffModeTransitionCommand, OwnershipGate, - decide, ownership_gate_requirement, ) -from ..goals.active_state_metadata import parse_state_frontmatter +from ..goals.active_state_metadata import parse_state_frontmatter, split_state_frontmatter from .contract import normalize_todo_claimed_by HANDOFF_MODE_SCHEMA_VERSION = "goal_handoff_mode_v0" @@ -281,122 +273,19 @@ def show_goal_handoff_mode( } -def _frontmatter_bounds(lines: list[str]) -> tuple[int, int]: - if not lines or lines[0].strip() != "---": +def _plan_legacy_mode(request: dict[str, Any]) -> dict[str, Any]: + from ..effect_runtime import effect_runtime_result + + result = effect_runtime_result("coordination.handoff_mode.legacy_plan", request) + if not isinstance(result, dict) or result.get("schema_version") != "loopx_legacy_handoff_mode_plan_result_v0": + raise HandoffModeError("invalid typed handoff plan", code="handoff_mode_plan_unavailable") + if result.get("outcome") == "rejected": raise HandoffModeError( - "active state file has no YAML front-matter; add one before setting " - "handoff_mode", - code="state_frontmatter_missing", + "handoff mode change rejected; resolve the reported state or ownership blockers", + code=result["code"], payload={key: value for key, value in result.items() + if key not in {"schema_version", "outcome", "code", "changed"}}, ) - for index in range(1, len(lines)): - if lines[index].strip() == "---": - return 1, index - raise HandoffModeError( - "active state front-matter is not terminated by ---", - code="state_frontmatter_missing", - ) - - -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. - - Event-projection overlays are intentionally outside this pre-NoKV scan. - Callers must not interpret an empty result as an authority-wide safety - guarantee; later governed writes still cross the selected handoff-mode - gate. - """ - - from ..work_items.task_lease import ( - lease_is_active, - read_lease, - runtime_root_from_registry, - task_lease_dir, - ) - from .active_state_todo_parser import parse_todo_source - from .todo_summary import structured_todo_item, todo_projection_sort_key - - # Quiescence needs normalized ownership facts, not status/resume/capability display. - # Retain the public offender order without evaluating unrelated projection rules. - claimed: list[dict[str, Any]] = [] - todos, _, sections = parse_todo_source(state_text) - for role in ("user", "agent"): - items = [structured_todo_item(item, role=role, source_section=sections[role]) for item in todos[role]] - for item in sorted(items, key=todo_projection_sort_key): - if not isinstance(item, dict) or item.get("done") is True: - continue - owner = normalize_todo_claimed_by(item.get("claimed_by")) - if owner: - claimed.append( - { - "todo_id": item.get("todo_id"), - "claimed_by": owner, - "status": item.get("status"), - } - ) - leases: list[dict[str, Any]] = [] - 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")): - lease = read_lease(path) - if lease_is_active(lease): - leases.append( - { - "todo_id": lease.get("todo_id"), - "owner": lease.get("owner"), - "expires_at": lease.get("expires_at"), - "lease_path": str(path), - } - ) - return claimed, leases - - -def _authority_offender_tokens( - offenders: list[dict[str, Any]], - *, - kind: str, -) -> tuple[str, ...]: - """Keep every offender represented without changing its public payload.""" - - return tuple( - str(offender.get("todo_id") or f"") - for index, offender in enumerate(offenders, start=1) - ) - - -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}") + return result def set_goal_handoff_mode( @@ -415,23 +304,13 @@ def set_goal_handoff_mode( Promoted Goals use one provider transaction; the remaining text below describes the unpromoted compatibility writer. - In v0, quiescence means no open todo materialized in the locked active-state - Markdown carries a claimed_by owner and no time-active lease file exists - under the goal. The scan does not overlay event-only todos, so success is a - pre-NoKV migration check rather than an authority-wide safety guarantee; - every later governed ownership or completion write still crosses the - selected mode's typed gate. Visible non-quiescence refuses with a typed - offender list and there is no force override. Hand-editing the front-matter - bypasses this check and is out of contract. + Changed modes require complete unclaimed Todo state and no time-active + leases. Event source locks remain held through the frontmatter replacement. + Hand-editing frontmatter bypasses this check and is outside the contract. """ - from ...file_lock import ( - exclusive_cross_runtime_file_lock as exclusive_file_lock, - ) - from ..work_items.task_lease import ( - runtime_root_from_registry, - task_lease_lock_path, - ) + from ..work_items.task_lease import runtime_root_from_registry + from .handoff_mode_source import handoff_mode_source from ..coordination.legacy_writer_fence import ( LegacyCoordinationWriterFenced, legacy_todo_write_transaction, @@ -493,97 +372,45 @@ def set_goal_handoff_mode( registry_path, goal_id, resolved_state_file, None, "handoff_mode_set", dry_run, runtime_root=runtime_root, ): - original = resolved_state_file.read_text(encoding="utf-8") - previous, previous_mode_fields = _previous_handoff_mode_fields( - parse_state_frontmatter(original).get(HANDOFF_MODE_FRONTMATTER_KEY) - ) + with resolved_state_file.open(encoding="utf-8", newline="") as source: + original = source.read() + metadata, body = split_state_frontmatter(original) + frontmatter = original[:len(original) - len(body)] + request = { + "schema_version": "loopx_legacy_handoff_mode_plan_request_v0", + "previous_value": metadata.get(HANDOFF_MODE_FRONTMATTER_KEY), + "requested_mode": requested, "frontmatter_text": frontmatter, + "todos": None, "leases": None, + } + plan = _plan_legacy_mode(request) payload = { - "ok": True, - "schema_version": HANDOFF_MODE_SCHEMA_VERSION, - "action": "set", - "goal_id": goal_id, - **previous_mode_fields, - "handoff_mode": requested, - "state_file": str(resolved_state_file), + "ok": True, "schema_version": HANDOFF_MODE_SCHEMA_VERSION, + "action": "set", "goal_id": goal_id, "state_file": str(resolved_state_file), + **{key: value for key, value in plan.items() if key.startswith("previous_mode")}, + "handoff_mode": requested, "changed": False, + **({"dry_run": True} if dry_run else {}), } - if previous == requested: - payload["changed"] = False - if dry_run: - payload["dry_run"] = True + if plan["outcome"] == "no_change": return payload - capture = None if dry_run else begin_todo_runtime_shadow_capture( - registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, - state_path=resolved_state_file, write_class="handoff_mode_set", - original_text=original, - ) - 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: - previous_core_mode = HandoffMode(previous) - else: - # Invalid persisted front-matter can be repaired, but it is - # never an idempotent transition. Pick any distinct typed - # source mode; quiescence is independent of the source mode. - previous_core_mode = next( - candidate - for candidate in HandoffMode - if candidate is not requested_core_mode - ) - transition = decide( - CoordinationSnapshot( - handoff_mode=previous_core_mode, - active_claimed_todo_ids=_authority_offender_tokens( - claimed, - kind="claimed", - ), - active_lease_todo_ids=_authority_offender_tokens( - leases, - kind="lease", - ), - ), - HandoffModeTransitionCommand(requested_mode=requested_core_mode), - ) - if transition.code == "handoff_mode_not_quiescent": - raise HandoffModeError( - "handoff_mode can only change while the goal is quiescent: " - f"{len(claimed)} claimed open todo(s), " - f"{len(leases)} time-active lease(s)", - code="handoff_mode_not_quiescent", - payload={ - "goal_id": goal_id, - "requested_mode": requested, - **previous_mode_fields, - "claimed_todos": claimed, - "active_leases": leases, - }, - ) - if transition.outcome is not DecisionOutcome.APPLY: - raise HandoffModeError( - f"handoff_mode transition rejected by authority core: " - f"{transition.code}", - code=transition.code, - payload={ - "goal_id": goal_id, - "requested_mode": requested, - **previous_mode_fields, - "claimed_todos": claimed, - "active_leases": leases, - }, - ) + if plan["outcome"] != "snapshot_required": + raise HandoffModeError("unexpected handoff planning phase", code="handoff_mode_plan_unavailable") + with handoff_mode_source(registry_path=registry_path, goal_id=goal_id, + state_path=resolved_state_file, state_text=original, runtime_root=runtime_root) as facts: + try: + plan = _plan_legacy_mode({**request, **facts}) + except HandoffModeError as error: + error.payload.update(goal_id=goal_id, requested_mode=requested) + raise + if plan.get("outcome") != "apply" or not isinstance(plan.get("next_frontmatter_text"), str): + raise HandoffModeError("incomplete handoff mutation plan", code="handoff_mode_plan_unavailable") if dry_run: - return {**payload, "dry_run": True, "changed": True} - lines = original.splitlines() - _write_handoff_mode_frontmatter(lines, requested) - new_text = "\n".join(lines) + ("\n" if original.endswith("\n") else "") + return {**payload, "changed": True} + capture = begin_todo_runtime_shadow_capture( + registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, + state_path=resolved_state_file, write_class="handoff_mode_set", original_text=original) write_captured_todo_state(capture, runtime_root=runtime_root, goal_id=goal_id, - state_path=resolved_state_file, text=new_text) + state_path=resolved_state_file, text=plan["next_frontmatter_text"] + body) + previous = str(plan["previous_mode"]) payload["changed"] = True from ..coordination.local_authority_shadow_observation import observe_local_authority_commit diff --git a/loopx/control_plane/todos/handoff_mode_source.py b/loopx/control_plane/todos/handoff_mode_source.py new file mode 100644 index 0000000000..23c04e0f01 --- /dev/null +++ b/loopx/control_plane/todos/handoff_mode_source.py @@ -0,0 +1,61 @@ +"""Locked compatibility inputs for the typed handoff transition planner. + +The active-state writer mutex is held by the caller. Event locks use the +append store's own lock and remain held until the mode write is durable. +""" +from __future__ import annotations + +from contextlib import ExitStack, contextmanager +from collections.abc import Iterator +from pathlib import Path +from typing import Any + + +def _event_paths(registry_path: Path, goal_id: str, state_path: Path) -> tuple[dict[str, Any], list[Path]]: + from ...history import load_registry + from ...registry import find_registry_goal + from ..status.active_state_projection import state_event_log_candidates + + goal = find_registry_goal(load_registry(registry_path), goal_id) or {"id": goal_id} + paths = state_event_log_candidates(goal, state_path=state_path) + return goal, sorted({path.expanduser().resolve() for path in paths}, key=str) + + +@contextmanager +def handoff_mode_source( + *, registry_path: Path, goal_id: str, state_path: Path, state_text: str, + runtime_root: Path, +) -> Iterator[dict[str, Any]]: + from ...event_sourced_state import AppendOnlyStateEventStore, StateEventError + from ...file_lock import exclusive_file_lock, exclusive_cross_runtime_file_lock + from ..runtime.time import now_local_iso + from ..work_items.task_lease import read_lease, task_lease_dir, task_lease_lock_path + from .goal_todo_projection import project_goal_todo_items + from .handoff_mode import HandoffModeError + + goal, paths = _event_paths(registry_path, goal_id, state_path) + with ExitStack() as stack: + # Same ordering for every mode writer; state -> event logs -> leases. + # Lock absent candidates too, so the first append cannot race the scan. + for path in paths: + stack.enter_context(exclusive_file_lock(path, operation="handoff_mode_set")) + stack.enter_context(exclusive_cross_runtime_file_lock( + task_lease_lock_path(runtime_root=runtime_root, goal_id=goal_id), operation="handoff_mode_set")) + try: + # Display projection can warn and fall back on malformed events. + # A safety decision must instead prove every candidate readable. + for path in paths: + AppendOnlyStateEventStore(path).load() + todos = project_goal_todo_items(goal, state_text=state_text, + state_path=state_path, rollout_events=[]) + except (OSError, StateEventError) as error: + raise HandoffModeError("cannot establish handoff quiescence from the event sources", + code="handoff_mode_source_unavailable") from error + fields = ("todo_id", "done", "status", "claimed_by", "archive_state") + leases = [] + for path in sorted(task_lease_dir(runtime_root=runtime_root, goal_id=goal_id).glob("todo_*.json")): + lease = read_lease(path) + if lease is not None: + leases.append({**lease, "lease_path": str(path)}) + yield {"todos": [{key: item[key] for key in fields if key in item} for item in todos], + "leases": leases, "observed_at": now_local_iso()} From 56864217a35b17c166d57f61e769b3afb37719db Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Tue, 22 Sep 2026 17:51:22 +0800 Subject: [PATCH 2/3] test(coordination): cover complete handoff sources and failed acknowledgements Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- tests/control_plane/test_goal_handoff_mode.py | 132 +++------------- .../control_plane/test_handoff_mode_source.py | 141 ++++++++++++++++++ .../handoff_mode_conformance.ts | 59 ++++++++ .../handoff_mode_plan.test.ts | 90 +++++++++++ 4 files changed, 307 insertions(+), 115 deletions(-) create mode 100644 tests/control_plane/test_handoff_mode_source.py create mode 100644 tests/control_plane_ts/handoff_mode_plan.test.ts diff --git a/tests/control_plane/test_goal_handoff_mode.py b/tests/control_plane/test_goal_handoff_mode.py index a2aa064f17..986ec21059 100644 --- a/tests/control_plane/test_goal_handoff_mode.py +++ b/tests/control_plane/test_goal_handoff_mode.py @@ -258,126 +258,28 @@ def test_goal_handoff_mode_without_frontmatter_is_legacy() -> None: assert goal_handoff_mode("## Agent Todo\n") == HANDOFF_MODE_LEGACY -def test_event_only_claim_is_not_a_v0_quiescence_offender_but_hard_gate_blocks_write( - tmp_path: Path, -) -> None: - """The v0 switch scan is materialized-state only, not a safety proof. - - An event-only claim is visible through the real todo projection but absent - from ACTIVE_GOAL_STATE.md, so the transition scan does not see it. After - the switch, the normal event writeback path must still cross the hard-lease - completion gate and leave both persisted surfaces unchanged on rejection. - """ - +def test_event_only_claim_blocks_mode_switch_without_mutating_either_source(tmp_path: Path) -> None: + """Unmaterialized claims are ownership facts, including during migration.""" registry, state = _write_workspace(tmp_path) event_log = state.with_name("events.jsonl") todo_id = "todo_event_only_claimed" - store = AppendOnlyStateEventStore(event_log) - store.append( - make_state_event( - event_id="evt-event-only-claimed", - goal_id=GOAL_ID, - event_type=TODO_ADDED, - refs={"todo_id": todo_id}, - payload={ - "role": "agent", - "title": "Complete the event-only claimed task.", - "task_class": "advancement_task", - "claimed_by": AGENT_A, - }, - recorded_at="2026-08-01T00:01:00+00:00", - producer="handoff-mode-event-only-characterization", - ) - ) - + AppendOnlyStateEventStore(event_log).append(make_state_event( + event_id="evt-event-only-claimed", goal_id=GOAL_ID, event_type=TODO_ADDED, + refs={"todo_id": todo_id}, payload={"role": "agent", "title": "Complete the event-only task.", + "task_class": "advancement_task", "claimed_by": AGENT_A}, + recorded_at="2026-08-01T00:01:00+00:00", producer="handoff-mode-regression")) assert todo_id not in state.read_text(encoding="utf-8") - projected = list_goal_todos( - registry_path=registry, - goal_id=GOAL_ID, - todo_id=todo_id, - ) - assert projected["source"] == "event_projection_with_markdown_overlay" - assert todo_id in projected["projection_overlay"]["event_only_todo_ids"] - assert projected["todo"]["status"] == "open" + projected = list_goal_todos(registry_path=registry, goal_id=GOAL_ID, todo_id=todo_id) assert projected["todo"]["claimed_by"] == AGENT_A - - event_log_before_switch = event_log.read_text(encoding="utf-8") - switched = set_goal_handoff_mode( - registry_path=registry, - goal_id=GOAL_ID, - mode=HANDOFF_MODE_HARD_LEASE, - ) - assert switched["changed"] is True - assert switched["handoff_mode"] == HANDOFF_MODE_HARD_LEASE - assert switched["previous_mode"] == HANDOFF_MODE_LEGACY - assert event_log.read_text(encoding="utf-8") == event_log_before_switch - - state_after_switch = state.read_text(encoding="utf-8") - event_log_after_switch = event_log.read_text(encoding="utf-8") - with pytest.raises(TaskLeaseError) as error: - complete_goal_todo( - registry_path=registry, - goal_id=GOAL_ID, - todo_id=todo_id, - claimed_by=AGENT_A, - agent_id=AGENT_A, - evidence="event-only completion still needs the post-switch lease gate", - no_followup=True, - ) - - assert error.value.code == "handoff_mode_requires_lease" - assert error.value.payload["handoff_mode"] == HANDOFF_MODE_HARD_LEASE - assert state.read_text(encoding="utf-8") == state_after_switch - assert event_log.read_text(encoding="utf-8") == event_log_after_switch - projected_after = list_goal_todos( - registry_path=registry, - goal_id=GOAL_ID, - todo_id=todo_id, - ) - assert projected_after["todo"]["status"] == "open" - assert projected_after["todo"]["claimed_by"] == AGENT_A - - -def test_mode_transition_keeps_each_missing_todo_id_offender( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, -) -> None: - registry, state = _write_workspace(tmp_path) - claimed = [ - {"todo_id": None, "claimed_by": AGENT_A, "status": "open"}, - {"todo_id": "", "claimed_by": AGENT_B, "status": "open"}, - ] - leases = [ - { - "todo_id": None, - "owner": AGENT_A, - "expires_at": "2026-08-01T01:00:00Z", - "lease_path": ".loopx/task-leases/lease-a.json", - }, - { - "todo_id": "", - "owner": AGENT_B, - "expires_at": "2026-08-01T01:00:00Z", - "lease_path": ".loopx/task-leases/lease-b.json", - }, - ] - monkeypatch.setattr( - "loopx.control_plane.todos.handoff_mode._quiescence_offenders", - lambda **_kwargs: (claimed, leases), - ) - before = state.read_bytes() - - with pytest.raises(HandoffModeError) as error: - set_goal_handoff_mode( - registry_path=registry, - goal_id=GOAL_ID, - mode=HANDOFF_MODE_HARD_LEASE, - ) - - assert error.value.code == "handoff_mode_not_quiescent" - assert error.value.payload["claimed_todos"] == claimed - assert error.value.payload["active_leases"] == leases - assert state.read_bytes() == before + before = state.read_bytes(), event_log.read_bytes() + for dry_run in (True, False): + with pytest.raises(HandoffModeError) as error: + set_goal_handoff_mode(registry_path=registry, goal_id=GOAL_ID, + mode=HANDOFF_MODE_HARD_LEASE, dry_run=dry_run) + assert error.value.code == "handoff_mode_not_quiescent" + assert error.value.payload["claimed_todos"] == [ + {"todo_id": todo_id, "claimed_by": AGENT_A, "status": "open"}] + assert (state.read_bytes(), event_log.read_bytes()) == before # --------------------------------------------------------------------------- diff --git a/tests/control_plane/test_handoff_mode_source.py b/tests/control_plane/test_handoff_mode_source.py new file mode 100644 index 0000000000..2dfb0e8f6e --- /dev/null +++ b/tests/control_plane/test_handoff_mode_source.py @@ -0,0 +1,141 @@ +"""Real compatibility sources: complete overlays and their mutation locks.""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from loopx.control_plane.todos.handoff_mode import HandoffModeError, set_goal_handoff_mode, show_goal_handoff_mode +from loopx.event_sourced_state import AppendOnlyStateEventStore, TODO_ADDED, make_state_event +from loopx.file_lock import LockAcquireTimeoutError, exclusive_file_lock + + +def workspace(root: Path) -> tuple[Path, Path, Path]: + state = root / "ACTIVE_GOAL_STATE.md" + state.write_text("---\ngoal_id: mode-source\nhandoff_mode: legacy\n---\n\n## Agent Todo\n", encoding="utf-8") + registry = root / "registry.json" + runtime = root / "runtime" + registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{"id": "mode-source", + "repo": str(root), "state_file": state.name, "adapter": {"kind": "harness_self_improvement"}}]})) + return registry, state, runtime + + +def append(path: Path, *, claimed: bool, todo_id: str = "todo_event") -> None: + AppendOnlyStateEventStore(path).append(make_state_event(event_id=f"add-{todo_id}", goal_id="mode-source", + event_type=TODO_ADDED, refs={"todo_id": todo_id}, payload={"role": "agent", "title": "Durable event task", + "task_class": "advancement_task", **({"claimed_by": "worker"} if claimed else {})}, + recorded_at="2026-09-20T00:00:00Z", producer="mode-source-fixture")) + + +def switch(registry: Path, **kwargs): + return set_goal_handoff_mode(registry_path=registry, goal_id="mode-source", mode="hard_lease", **kwargs) + + +@pytest.mark.parametrize("registered", [False, True]) +def test_unmaterialized_claim_at_end_of_large_event_source_blocks(tmp_path: Path, registered: bool) -> None: + registry, state, _ = workspace(tmp_path) + path = tmp_path / ("declared-events.jsonl" if registered else "events.jsonl") + if registered: + value = json.loads(registry.read_text()) + value["goals"][0]["state_event_log"] = str(path) + registry.write_text(json.dumps(value)) + events = [make_state_event(event_id=f"add-{i}", goal_id="mode-source", event_type=TODO_ADDED, + refs={"todo_id": f"todo_event_{i}"}, payload={"role": "user" if i % 2 else "agent", + "title": f"Task {i}", "task_class": "advancement_task", **({"claimed_by": "worker"} if i == 520 else {})}, + recorded_at="2026-09-20T00:00:00Z", producer="mode-source-fixture") for i in range(521)] + AppendOnlyStateEventStore(path).append_many(events) + before = state.read_bytes(), path.read_bytes() + for dry_run in (True, False): + with pytest.raises(HandoffModeError) as caught: + switch(registry, dry_run=dry_run) + assert caught.value.code == "handoff_mode_not_quiescent" + assert [row["todo_id"] for row in caught.value.payload["claimed_todos"]] == ["todo_event_520"] + assert (state.read_bytes(), path.read_bytes()) == before + + +def test_unclaimed_overlay_stays_unmaterialized_and_locked_until_durable_write(tmp_path: Path, monkeypatch) -> None: + registry, state, _ = workspace(tmp_path) + path = state.with_name("events.jsonl") + append(path, claimed=False) + from loopx.control_plane.coordination import runtime_shadow_writer_adapter as adapter + original_write = adapter.write_captured_todo_state + seen = [] + + def locked_write(*args, **kwargs): + with pytest.raises(LockAcquireTimeoutError): + with exclusive_file_lock(path, timeout_seconds=0): + pytest.fail("event append lock was released before mode write") + seen.append(True) + return original_write(*args, **kwargs) + + monkeypatch.setattr(adapter, "write_captured_todo_state", locked_write) + before = path.read_bytes() + result = switch(registry) + assert result["changed"] is True and seen == [True] + assert path.read_bytes() == before + assert "todo_event" not in state.read_text() + assert show_goal_handoff_mode(registry_path=registry, goal_id="mode-source")["handoff_mode"] == "hard_lease" + with exclusive_file_lock(path, timeout_seconds=0): + pass + + +@pytest.mark.parametrize("contents", ["not json\n", '{"schema_version":"unknown"}\n']) +def test_unreadable_event_source_is_not_quiescence(tmp_path: Path, contents: str) -> None: + registry, state, _ = workspace(tmp_path) + path = state.with_name("events.jsonl") + path.write_text(contents) + before = state.read_bytes() + with pytest.raises(HandoffModeError) as caught: + switch(registry) + assert caught.value.code == "handoff_mode_source_unavailable" + assert state.read_bytes() == before + + +def test_invalid_unused_event_candidate_does_not_fall_back_silently(tmp_path: Path) -> None: + registry, state, _ = workspace(tmp_path) + declared = tmp_path / "declared.jsonl" + append(declared, claimed=False) + value = json.loads(registry.read_text()) + value["goals"][0]["state_event_log"] = str(declared) + registry.write_text(json.dumps(value)) + state.with_name("events.jsonl").write_text("corrupt fallback\n") + with pytest.raises(HandoffModeError) as caught: + switch(registry) + assert caught.value.code == "handoff_mode_source_unavailable" + + +def test_noop_does_not_need_to_repair_or_scan_source(tmp_path: Path) -> None: + registry, state, _ = workspace(tmp_path) + state.write_text(state.read_text().replace("legacy", "hard_lease")) + state.with_name("events.jsonl").write_text("corrupt unrelated event\n") + before = state.read_bytes() + assert switch(registry)["changed"] is False + assert state.read_bytes() == before + + +@pytest.mark.parametrize("ending", [b"", b"\r\n"]) +def test_public_mode_write_preserves_unrelated_bytes(tmp_path: Path, ending: bytes) -> None: + registry, state, _ = workspace(tmp_path) + source = '---\r\ngoal_id: mode-source\r\ntitle: "a\u2028b"\r\nhandoff_mode: legacy\r\n---\r\n\r\n## Agent Todo'.encode() + ending + state.write_bytes(source) + assert switch(registry)["changed"] is True + assert state.read_bytes() == source.replace(b"handoff_mode: legacy", b"handoff_mode: hard_lease") + + +def test_duplicate_mode_fields_reject_without_partial_repair(tmp_path: Path) -> None: + registry, state, _ = workspace(tmp_path) + state.write_text(state.read_text().replace("handoff_mode: legacy", "handoff_mode: legacy\nhandoff_mode: banana")) + before = state.read_bytes() + with pytest.raises(HandoffModeError) as caught: + switch(registry) + assert caught.value.code == "handoff_mode_duplicate_field" + assert state.read_bytes() == before + + +def test_large_prose_never_crosses_the_mode_plan_transport(tmp_path: Path) -> None: + registry, state, _ = workspace(tmp_path) + original = state.read_bytes() + b"\n## Evidence\n" + b"Unrelated durable prose.\n" * 150_000 + state.write_bytes(original) + assert switch(registry)["changed"] is True + assert state.read_bytes() == original.replace(b"handoff_mode: legacy", b"handoff_mode: hard_lease") diff --git a/tests/control_plane_ts/handoff_mode_conformance.ts b/tests/control_plane_ts/handoff_mode_conformance.ts index 0eccfeda9a..a1bd13ae2e 100644 --- a/tests/control_plane_ts/handoff_mode_conformance.ts +++ b/tests/control_plane_ts/handoff_mode_conformance.ts @@ -133,4 +133,63 @@ export function registerHandoffModeConformance(provider: string, factory: Author assert.equal((await executeHandoffModeSet(store, request)).status, "replayed"); assert.deepEqual(await head(store), after); }); + test(`${provider}: thrown post-commit response is recovered and never retried as a fresh write`, async t => { + const {store} = await factory(t); + await seed(store); + let commits = 0; + const lost = intercept(store, async commit => { + commits++; + assert.equal((await store.commitAuthority(commit)).status, "applied"); + throw new Error("transport disconnected after commit"); + }); + assert.equal((await executeHandoffModeSet(lost, request)).status, "recovered"); + assert.equal((await executeHandoffModeSet(lost, request)).status, "replayed"); + assert.equal(commits, 1); + }); + + test(`${provider}: unobserved receipt after commit requires same-operation recovery`, async t => { + const {store} = await factory(t); + await seed(store); + let committed = false; + const unavailable = intercept(store, async commit => { + const result = await store.commitAuthority(commit); + committed = true; + return result; + }); + unavailable.readReceipt = id => { + if (committed) throw new Error("receipt connection unavailable"); + return store.readReceipt(id); + }; + const result = await executeHandoffModeSet(unavailable, request); + assert.equal(result.status, "ambiguous"); + assert.equal(result.reason_code, "coordination_receipt_recovery_required"); + assert.deepEqual(result.recovery, {operation_id: request.operation_id, retry_with_same_operation_id: true}); + const after = await head(store); + assert.equal((await executeHandoffModeSet(store, request)).status, "replayed"); + assert.deepEqual(await head(store), after); + }); + + for (const corruption of ["changed", "previous_mode", "operation_id", "handoff_mode"] as const) { + test(`${provider}: malformed historical ${corruption} decision cannot masquerade as a no-op`, async t => { + const {store} = await factory(t); + await seed(store); + await executeHandoffModeSet(store, request); + const before = await head(store); + const corrupt = intercept(store, commit => store.commitAuthority(commit)); + corrupt.readReceipt = async id => { + const receipt = await store.readReceipt(id); + if (receipt.status !== "found") return receipt; + const copied = structuredClone(receipt); + const decision = copied.receipts[0]!.decision as JsonObject; + if (corruption === "changed") delete decision.changed; + else decision[corruption] = "wrong"; + return copied; + }; + const result = await executeHandoffModeSet(corrupt, request); + assert.equal(result.reason_code, "invalid_coordination_command_receipt"); + assert.equal(result.changed, false); + assert.deepEqual(await head(store), before); + }); + } + } diff --git a/tests/control_plane_ts/handoff_mode_plan.test.ts b/tests/control_plane_ts/handoff_mode_plan.test.ts new file mode 100644 index 0000000000..4bcc5c10e2 --- /dev/null +++ b/tests/control_plane_ts/handoff_mode_plan.test.ts @@ -0,0 +1,90 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {HANDOFF_MODES, HANDOFF_MODE_PLAN_SCHEMA, planHandoffMode} from "../../loopx/control_plane/coordination/handoff_mode_policy.ts"; +import {handoffQuiescence} from "../../loopx/control_plane/coordination/handoff_mode_facts.ts"; +import {LEGACY_HANDOFF_PLAN_SCHEMA, planLegacyHandoffMode, patchHandoffMode} from "../../loopx/control_plane/coordination/handoff_mode_legacy_plan.ts"; + +const at = "2026-09-20T00:00:00Z"; +const claim = {todo_id: "todo_owned", claimed_by: "worker", done: false, status: "open", archive_state: "active"}; +const lease = {schema_version: "task_lease_v0", todo_id: "todo_leased", owner: "worker", + status: "active", expires_at: "2026-09-20T00:00:01Z"}; +const request = {schema_version: LEGACY_HANDOFF_PLAN_SCHEMA, frontmatter_text: "---\n---\n## Agent Todo\n", + previous_value: null, requested_mode: "hard_lease", todos: [], leases: [], observed_at: at}; + +for (const previous of HANDOFF_MODES) for (const requested of HANDOFF_MODES) { + for (const blockers of [false, true]) test(`${previous} -> ${requested}, blockers=${blockers}: legacy/canonical decision parity`, () => { + const expected = previous === requested ? "no_change" : blockers ? "rejected" : "apply"; + const legacy = planLegacyHandoffMode({...request, previous_value: previous, requested_mode: requested, + todos: blockers ? [claim] : [], leases: blockers ? [lease] : []}); + const canonical = planHandoffMode({schema_version: HANDOFF_MODE_PLAN_SCHEMA, + previous_mode: previous, requested_mode: requested, + active_claimed_todo_ids: blockers ? [claim.todo_id] : [], active_lease_todo_ids: blockers ? [lease.todo_id] : []}); + assert.equal(legacy.outcome, expected); + assert.equal(canonical.outcome, expected); + assert.equal(legacy.changed, expected === "apply"); + }); +} + +test("invalid persisted mode is an explicit repair; never a fabricated previous valid mode", () => { + for (const mode of HANDOFF_MODES) { + const requestWithInvalid = {...request, previous_value: "banana", requested_mode: mode}; + const allowed = planLegacyHandoffMode(requestWithInvalid); + assert.equal(allowed.outcome, "apply"); + assert.equal(allowed.previous_mode, "banana"); + assert.equal(allowed.previous_mode_valid, false); + assert.equal(allowed.previous_mode_error_code, "invalid_handoff_mode"); + for (const facts of [{todos: [claim]}, {leases: [lease]}]) { + assert.equal(planLegacyHandoffMode({...requestWithInvalid, ...facts}).outcome, "rejected"); + } + } +}); + +test("no-op needs no ownership snapshot, but a changed intent explicitly requires one", () => { + assert.equal(planLegacyHandoffMode({...request, todos: null, leases: null}).outcome, "snapshot_required"); + const result = planLegacyHandoffMode({...request, previous_value: "hard_lease", + frontmatter_text: "no metadata", todos: null, leases: null}); + assert.equal(result.outcome, "no_change"); + assert.equal(result.changed, false); + assert.equal(result.next_frontmatter_text, undefined); +}); + +test("complete claim/lease facts retain every blocker including missing identities", () => { + const result = handoffQuiescence([claim, {...claim, todo_id: null}, {...claim, done: true}, + {...claim, archive_state: "archive"}, {...claim, claimed_by: " "}], + [lease, {...lease, todo_id: null}, {...lease, status: "released"}, {...lease, expires_at: at}], at); + assert.deepEqual(result.claimed_todos.map(row => row.todo_id), ["todo_owned", null]); + assert.deepEqual(result.active_leases.map(row => row.todo_id), ["todo_leased", null]); + assert.throws(() => handoffQuiescence([], [{...lease, schema_version: "future"}], at), /schema/); + assert.throws(() => handoffQuiescence([], [{...lease, expires_at: "garbage"}], at)); + assert.throws(() => handoffQuiescence([], [], "tomorrow")); + assert.throws(() => handoffQuiescence([{...claim, claimed_by: 123}], [], at)); +}); + +test("frontmatter edit retains CRLF, quoted Unicode separators, body and final-newline choice", () => { + for (const newline of ["\n", "\r\n"]) for (const trailing of ["", newline]) { + const prefix = `---${newline}title: "a\u2028b\u0085c"${newline}`; + const suffix = `---${newline}body: handoff_mode: legacy${trailing}`; + const input = `${prefix}handoff_mode: legacy${newline}${suffix}`; + assert.equal(patchHandoffMode(input, "hard_lease"), `${prefix}handoff_mode: hard_lease${newline}${suffix}`); + assert.equal(patchHandoffMode(prefix + suffix, "hard_lease"), `${prefix}handoff_mode: hard_lease${newline}${suffix}`); + } +}); + +test("missing frontmatter and duplicate fields cannot produce a partial repair", () => { + for (const [text, code] of [["no frontmatter", "state_frontmatter_missing"], + ["---\nhandoff_mode: legacy\nhandoff_mode: banana\n---\n", "handoff_mode_duplicate_field"]]) { + const plan = planLegacyHandoffMode({...request, frontmatter_text: text, previous_value: "banana"}); + assert.equal(plan.outcome, "rejected"); + assert.equal(plan.code, code); + assert.equal(plan.next_frontmatter_text, undefined); + } +}); + + +test("Unicode separators cannot invent metadata keys or close the frontmatter", () => { + for (const separator of ["\u2028", "\u2029"]) { + const prefix = `---\ntitle: "first${separator}handoff_mode: banana${separator}---"\n`; + const source = prefix + "handoff_mode: legacy\n---\n"; + assert.equal(patchHandoffMode(source, "hard_lease"), prefix + "handoff_mode: hard_lease\n---\n"); + } +}); From 3ec1e1f08c70756f9e5ce588e2ecdd3400713beb Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Tue, 22 Sep 2026 17:51:30 +0800 Subject: [PATCH 3/3] docs(coordination): disclose complete handoff transition checks Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- ...shared-goal-authority-state-provider-v0.md | 9 ++++ .../typescript-control-plane-migration-v0.md | 10 ++++ ...script-control-plane-migration-v0.zh-CN.md | 7 +++ docs/reference/handoff-mode.md | 47 +++++++++++++++++-- 4 files changed, 68 insertions(+), 5 deletions(-) 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 0fe99fd3e6..e8369e173d 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -31,6 +31,15 @@ ## Current implementation checkpoint +Handoff-mode changes now share one TS ownership-fact classifier before and +after promotion. Legacy event-only claims reject rather than disappear at a +Markdown boundary; event append locks protect the observation through writeback. +Canonical changes reuse durable command receipt recovery. This is an L2/L3 +compatibility correction with Python decision deletion, not cohort migration, +SQLite D2 completion or a default flip. The remaining 5–8 packages still depend +on executor/consumer closure, qualification, integrated migration and onboarding. +[Operation, repair and recovery](../../reference/handoff-mode.md). + The terminal caller family now binds review and validation to the canonical source and recovers historical receipts independently of private argv. Agent completion and Monitor stop share current-head display acknowledgement with diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 24cb668ddc..562e9003e4 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -22,6 +22,16 @@ Retain T0 caller/parity inventory, T1/T2 transaction/effect convergence, T3 comp ## Current implementation checkpoint +Handoff-mode transition now shares typed ownership facts and an explicit +valid/invalid previous-mode state across legacy and canonical paths. Python's +blocker classification, artificial previous mode and whole-text rewrite are +removed; its retained boundary is source projection/locking and capture IO. +The legacy scan includes event-only claims, and canonical mode receipts reuse +command recovery with strict historical decisions. Full-source snapshot and +real-provider validation guard this T1/T2 replacement. This closes a rule and +caller discrepancy, not a whole default-cutover package; the conditional 5–8 +package estimate remains. [Changed behavior and recovery](../../reference/handoff-mode.md). + Terminal review and validation now converge in the existing TS terminal owner. Agent completion and Monitor stop reuse Chat's canonical receipt-first recovery and display acknowledgement; v2 binds validation continuation to its source 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 c5dc54fa3c..fe75c840bd 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 @@ -21,6 +21,13 @@ ## 当前实现检查点 +模式切换的旧路径与 canonical 路径现在共用 TS 所有权事实及显式的有效/无效旧模式。 +删除 Python 的阻塞分类、伪造旧模式和整篇文本重写，保留来源投影、锁与 capture IO。 +旧扫描补齐事件独有 claim，canonical 回执复用 command recovery 并严格校验历史决策。 +完整快照和真实 provider 验证覆盖这一 T1/T2 替换；它关闭一处规则/调用差异， +不代表关闭整项默认切换交付包，条件性的 5–8 包估算不变。 +[行为变化与恢复](../../reference/handoff-mode.md)。 + 终结审核与验证已收敛到既有 TS terminal owner：Agent 完成、Monitor 停止复用 Chat 先恢复 canonical 回执再确认显示的路径；v2 把验证 continuation 绑定来源 revision， 准入/回放之后才请求私有声明。删除 Python 的终结操作审核分流和提前解析声明编排。 diff --git a/docs/reference/handoff-mode.md b/docs/reference/handoff-mode.md index 67fdbae3f9..82a830fda2 100644 --- a/docs/reference/handoff-mode.md +++ b/docs/reference/handoff-mode.md @@ -14,7 +14,7 @@ loopx handoff-mode set --goal-id example-goal --mode soft_claim --format json ``` Before promotion, these commands use the existing frontmatter writer and its -state/lease locks. After promotion, they use the selected canonical provider; +state/event/lease locks. After promotion, they use the selected canonical provider; `show` returns `source=canonical_provider` and its `provider_revision`, even if Markdown is stale or missing. `--runtime-root` applies to both show and set. Provider errors fail closed. A leftover local lease file cannot override an @@ -28,9 +28,28 @@ cannot prove quiescence. Concurrent mutations invalidate the CAS snapshot and return a conflict without switching the mode. Todos, lease records and their read-model digests are preserved by the mode change. -The unpromoted scan retains its older materialized-state scope: it does not -claim to include event-only Todos. Its quiescence decision and the canonical -transaction now share one typed policy. No default mode changes. +The unpromoted scan now includes the same complete event-overlay Todo view as +Todo listing, without its display limit. This changes the previous behavior: +an event-only claim now rejects a mode switch. Every configured/fallback event +candidate must be readable; corrupt input returns `handoff_mode_source_unavailable` +instead of silently falling back to apparently empty Markdown. The append store +locks (including absent candidate paths) remain held through the durable mode +write, followed by the existing per-goal lease mutex. Direct unsupported file +edits are outside this contract. + +Both paths use the same TS claim/lease classifier and mode-transition rule. +An identical valid mode remains a no-op even with active work. A malformed +legacy mode can be repaired only when quiescent; the result retains its actual +`previous_mode` and `previous_mode_valid=false`. Duplicate mode fields reject +with `handoff_mode_duplicate_field`, and missing frontmatter rejects a changed +mode with `state_frontmatter_missing`. Canonical malformed state still rejects; +legacy repair does not grant permission to repair a canonical head. + +Only frontmatter and compact ownership facts enter the legacy TS plan. Python +keeps the source locks, event projection and existing capture/writeback adapter; +the body never crosses the mode-plan transport. The scalar replacement preserves +unrelated metadata, CRLF/LF, Unicode separators and the final newline. No default +mode, provider or capability changes. ## Preserve claims during authority promotion @@ -97,6 +116,12 @@ A retry's clock may advance; it still recovers the original result. Even an accepted unchanged canonical set seals a receipt and advances provider revision, while returning `changed=false`. If another mode was selected afterward, replay returns the original decision without restoring it. Use `show` for current mode. +A thrown commit response follows the same durable receipt recovery as other +canonical commands. If the write may have committed but the receipt cannot be +read, the result is `ambiguous` with `coordination_receipt_recovery_required`: +retry the same operation ID. A malformed historical decision is rejected as +`invalid_coordination_command_receipt`, not coerced into an unchanged success. + Preview writes neither a mode nor an operation receipt. `--operation-id` requires canonical authority; the legacy writer does not promise durable operation replay. @@ -120,7 +145,19 @@ provider 失败明确报错，不回退旧文件。现有 Todo-section 投影不 切换要求完整快照内不存在未完成的已认领活动 Todo、不存在有效 lease。过期时间 恰好等于观察时间视为已过期；非法有效期或未知 lease schema 不能作为空闲证据。 并发修改使 CAS 冲突，不能在旧检查结果上继续切换。原 Todo、lease 和摘要不变。 -未晋升路径仍仅扫描物化状态，不宣称覆盖 event-only Todo；两条路径共用 TS 切换规则。 +未晋升路径现在也读取完整事件覆盖视图，包含显示分页之外的 Todo。因此旧行为发生改变： +仅在事件中存在的 claim 也会阻止切换。所有事件候选源必须可读，损坏源返回 +`handoff_mode_source_unavailable`，不能回退 Markdown 后宣称空闲。事件追加锁从读取 +保持到模式写回完成，再配合已有 lease 锁；直接手改文件仍不在该合同内。 + +两条路径共用 TS 所有权分类和切换规则。相同合法模式仍是 no-op；非法旧模式以显式 +无效状态进入修复，只允许在无在途工作时修复，不再伪造另一个合法旧模式。 +重复字段拒绝为 `handoff_mode_duplicate_field`，缺少 frontmatter 时拒绝变更。 +只把 frontmatter 和必要事实传给 TS，Python 保留锁、事件投影和 capture 适配； +正文不进入计划传输，并保留 CRLF、Unicode 分隔符和末尾换行。默认模式和 provider 不变。 + +canonical 提交响应丢失时复用已有回执恢复；若回执暂时不可读，返回 ambiguous 并要求 +以同一个 operation ID 重试。损坏的历史决策明确拒绝，不能当作“成功但没变化”。 对于无法清空活跃 claim 的 Goal，整 Goal authority 晋升提供一个更窄的显式迁移入口： `--handoff-mode-migration preserve` 只切换存储权威并保留 `legacy`／`soft_claim`；