From 5d2246e3e13c697b665c470ac6deef4c6cd11336 Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Sun, 25 Oct 2026 22:43:15 +0800 Subject: [PATCH] refactor(delegation): collapse turn recovery bridge into TypeScript Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- loopx/collaboration_mcp.py | 22 ++- .../control_plane/collaboration/delegation.ts | 41 +++++- .../collaboration/delegation_validation.py | 127 +++++------------- .../control_plane/effect_runtime_handlers.ts | 4 +- .../turn_driver/turn_journal_query.ts | 48 ++++++- .../turn_journal_query.test.ts | 29 ++++ 6 files changed, 162 insertions(+), 109 deletions(-) diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py index e3a6ddd257..1b6887272f 100644 --- a/loopx/collaboration_mcp.py +++ b/loopx/collaboration_mcp.py @@ -37,10 +37,7 @@ ) from .control_plane.coordination.local_authority import local_authority_is_promoted from .control_plane.runtime.validation_command import noninteractive_validation -from .control_plane.turn_driver.journal_store import ( - load_turn_journal, - turn_journal_path, -) +from .control_plane.turn_driver.journal_store import turn_journal_path from .control_plane.turn_driver.host_binding import turn_host_arg_option from .control_plane.turn_driver.host_process_transport import ( HOST_PROCESS_DRAINING, HOST_PROCESS_RECORD_ENV, @@ -812,11 +809,13 @@ def revalidate(self, operation_id: str) -> dict: binding = self._bound(row, require_active=True) if row["status"] != "rejected": raise ValueError("revalidation requires the original independent task failure") - failure = delegation_validation.task_failure(self, row, binding) + inspection = delegation_validation.turn_recovery(self, row, binding) + failure = inspection["task_failure"] if failure is None: # A previous recheck may have committed before its response or # projection write arrived. Reconcile that same execution first. - recovered = delegation_validation.recover_validated_settlement(self, path, row, binding) + recovered = delegation_validation.recover_validated_settlement( + self, path, row, binding, inspection=inspection) if not recovered: raise ValueError("revalidation requires the original independent task failure") else: @@ -900,9 +899,10 @@ def _read_current(self, operation_id: str) -> dict: if row.get("error"): result["error"] = row["error"] if row["status"] == "rejected": - if not active and stop is None and delegation_validation.committed_revalidation_key(self, row, binding) is not None: + inspection = delegation_validation.turn_recovery(self, row, binding) + if not active and stop is None and inspection["committed_revalidation"]: result["recovery_required"] = True - failure = delegation_validation.task_failure(self, row, binding) + failure = inspection["task_failure"] if failure is not None: result["task_failure"] = failure return result @@ -1021,11 +1021,7 @@ def _worker_lane_released(self, row: dict, stop: dict, binding: dict) -> tuple[b return False, state def _turn_journal_status(self, row: dict, binding: dict) -> str | None: - turn_key = row.get("turn_key") or delegation_validation.matching_turn_key(self, row, binding) - if not turn_key: - return None - journal = load_turn_journal(turn_journal_path(self.root, goal_id=self.goal_id, turn_key=turn_key)) - status = journal.get("status") if isinstance(journal, dict) else None + status = delegation_validation.turn_recovery(self, row, binding)["status"] return str(status) if status else None def _new_stop_record(self, row: dict, *, requested_by: str, worker: dict | None) -> dict: diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts index 3a5c016a82..a25b166d75 100644 --- a/loopx/control_plane/collaboration/delegation.ts +++ b/loopx/control_plane/collaboration/delegation.ts @@ -1,7 +1,7 @@ /** Explicit local execution bindings. Registration/messages alone grant no launch. * These are host observations; canonical task/Turn/acceptance remain authoritative. */ import type {JsonObject} from "../effect_program.ts"; -import {requireInteger, requireJsonObject, requireStringArray} from "../runtime_decode.ts"; +import {jsonObject, requireInteger, requireJsonObject, requireStringArray} from "../runtime_decode.ts"; import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; import {canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts"; import {acceptanceValidationEffects, goalAcceptanceTodoDigest, type AcceptanceCompletionRequirements} from "../goals/acceptance_contract.ts"; @@ -10,6 +10,7 @@ import {readTurnSelectionRejection, turnSelectionRejectionState} from "../turn_d import { BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; import {PROJECTION_ENVELOPE_SEAL_REQUEST, sealProjectionEnvelope} from "../projection_envelope.ts"; import {normalizeCollaborationBrief, planCollaborationRequestPublication} from "./semantic_request.ts"; +import {findDelegationTurnJournal} from "../turn_driver/turn_journal_query.ts"; function requireThat(ok: unknown, message: string): asserts ok { if (!ok) throw new EffectRuntimeRequestError(message); @@ -18,6 +19,44 @@ function text(value: unknown): value is string { return typeof value === "string" && value.length > 0 && value.length <= 4096; } +function revalidatedStatus(params: JsonObject, turnKey: unknown): JsonObject { + if (params.revalidated_result == null) return {}; + const receipt = requireJsonObject(params.revalidated_result, "revalidated Turn receipt"); + const lineage = jsonObject(receipt.lineage); + const intent = jsonObject(params.validation_failure); + return {revalidated_status: recoverRevalidatedDelegation({ + from: params.status, + original_task_failure: intent?.stage === "task_postcondition" + && intent.turn_key === receipt.turn_key && receipt.turn_key === turnKey + && lineage?.goal_id === params.goal_id && lineage?.agent_id === params.agent_id + && lineage?.todo_id === params.todo_id, + committed_progress: receipt.result_kind === "validated_progress", + host_reinvoked: receipt.host_reinvoked, + }).status}; +} + +export async function delegationTurnRecovery(params: JsonObject): Promise { + const evidence = await findDelegationTurnJournal(params); + const recovery = jsonObject(evidence.recovery); + const turnKey = evidence.turn_key; + return { + turn_key: turnKey, + task_failure: recovery?.task_failure ?? null, + committed_revalidation: jsonObject(params.validation_failure)?.turn_key === turnKey + && recovery?.status === "committed" && recovery?.result_kind === "validated_progress", + validated_settlement: recovery?.validated_settlement === true, + completed_phases: recovery?.completed_phases ?? null, + task_validation_ok: recovery?.task_validation_ok === true, + ...(recovery?.task_validation_ok === true ? {task_validation: {ok: true}} : {}), + ...(recovery?.host_result_turn_key === turnKey ? {host_result: { + turn_key: turnKey, result_kind: recovery?.host_result_kind, + }} : {}), + status: recovery?.status ?? null, + result_kind: recovery?.result_kind ?? null, + ...revalidatedStatus(params, turnKey), + }; +} + /** The canonical acceptance reader resolves owner scope before this plan. * A null requirement is out of scope/disabled, never an unbound-task fallback. * Private declarations must match the current Todo authority, also on readback. */ diff --git a/loopx/control_plane/collaboration/delegation_validation.py b/loopx/control_plane/collaboration/delegation_validation.py index ee4c0c843a..91e0aa2283 100644 --- a/loopx/control_plane/collaboration/delegation_validation.py +++ b/loopx/control_plane/collaboration/delegation_validation.py @@ -14,9 +14,6 @@ from ..agents.workspace_guard import capture_delivery_workspace from ..effect_runtime import effect_runtime_result from . import delegation_results -from ..turn_driver.journal_store import ( - find_loopx_turn_key_by_settlement_identity, load_turn_journal, turn_journal_path, -) from ..goals.acceptance import ( inspect_goal_acceptance, run_goal_acceptance_validation_effect, @@ -84,98 +81,52 @@ def validate(service, binding: dict) -> dict: return after -def task_failure(service, row, binding): - turn_key = matching_turn_key(service, row, binding) - if not turn_key: - return None - journal = load_turn_journal(turn_journal_path(service.root, goal_id=service.goal_id, turn_key=turn_key)) - if not journal or journal.get("result_kind") != "validation_failed": - return None - return effect_runtime_result("turn.task_validation_failure", { - "result_kind": journal.get("result_kind"), "status": journal.get("status"), - "receipt": journal.get("receipt"), "validation": journal.get("task_validation"), - **({"validation_stage": journal["validation_stage"]} if "validation_stage" in journal else {}), - })["failure"] - - -def committed_revalidation_key(service, row, binding): - """A retained recheck intent can outlive the original failed journal.""" - intent = row.get("validation_failure") - if not isinstance(intent, dict): - return None - turn_key = matching_turn_key(service, row, binding) - if not turn_key or intent.get("turn_key") != turn_key: - return None - journal = load_turn_journal(turn_journal_path(service.root, goal_id=service.goal_id, turn_key=turn_key)) - if journal and journal.get("status") == "committed" and journal.get("result_kind") == "validated_progress": - return turn_key - return None - - def record_revalidation_result(service, path, row, binding, result): from ..turn_driver.loop_controller import ValidatedTurnReceipt receipt = ValidatedTurnReceipt.from_execution(result) - intent = row.get("validation_failure") or {} - decision = effect_runtime_result("collaboration.delegation.revalidated", { - "from": row["status"], - "original_task_failure": intent.get("stage") == "task_postcondition" - and intent.get("turn_key") == receipt.turn_key - and receipt.turn_key == matching_turn_key(service, row, binding) - and receipt.lineage == {"goal_id": service.goal_id, "agent_id": binding["agent_id"], "todo_id": binding["todo_id"]}, - "committed_progress": receipt.result_kind.value == "validated_progress", + decision = turn_recovery(service, row, binding, revalidated_result={ + "turn_key": receipt.turn_key, "result_kind": receipt.result_kind.value, + "lineage": dict(receipt.lineage), "host_reinvoked": result.get("effects", {}).get("host_invoked"), }) + if decision["revalidated_status"] != "turn_returned": + raise ValueError("revalidation recovery requires the original committed Turn") service._record_turn_result(path, row, result, publish=False) - row["status"] = decision["status"] + row["status"] = decision["revalidated_status"] row.pop("error", None) service._fenced_write(path, row) +def turn_recovery(service, row, binding, *, revalidated_result=None): + return effect_runtime_result("collaboration.delegation.turn_recovery", { + "runtime_root": str(service.root.resolve()), "goal_id": service.goal_id, + "agent_id": binding["agent_id"], "todo_id": binding["todo_id"], + "turn_instance_id": service._turn_instance_id(row), "status": row.get("status"), + "validation_failure": row.get("validation_failure"), + "revalidated_result": revalidated_result, + }) + + def matching_turn_key(service, row: dict, binding: dict) -> str | None: - """Find only the journal bound to this operation's settlement identity.""" - - return find_loopx_turn_key_by_settlement_identity( - service.root, - goal_id=service.goal_id, - agent_id=binding["agent_id"], - todo_id=binding["todo_id"], - turn_instance_id=service._turn_instance_id(row), - ) + return turn_recovery(service, row, binding)["turn_key"] def validated_turn_journal(service, row: dict, binding: dict) -> dict | None: - turn_key = matching_turn_key(service, row, binding) - if turn_key is None: + inspection = turn_recovery(service, row, binding) + if not inspection["validated_settlement"]: return None - journal = load_turn_journal( - turn_journal_path(service.root, goal_id=service.goal_id, turn_key=turn_key) - ) - if journal is None: - return None - phases = journal.get("completed_phases") - validation = journal.get("task_validation") - host_result = journal.get("host_result") - if ( - journal.get("status") != "in_progress" - or journal.get("result_kind") != "validated_progress" - or phases != ["host_execute", "typed_result", "validation"] - or not isinstance(validation, dict) - or validation.get("ok") is not True - or not isinstance(host_result, dict) - or host_result.get("turn_key") != turn_key - or host_result.get("result_kind") != "validated_progress" - ): - return None - row["turn_key"] = turn_key - return journal + row["turn_key"] = inspection["turn_key"] + return {"status": inspection["status"], "result_kind": inspection["result_kind"], + "task_validation": inspection["task_validation"], "host_result": inspection["host_result"]} def recover_validated_settlement( - service, path: Path, row: dict, binding: dict + service, path: Path, row: dict, binding: dict, *, inspection: dict | None = None ) -> bool: """Reopen only an exact, independently validated settlement boundary.""" - committed_key = committed_revalidation_key(service, row, binding) + inspection = inspection or turn_recovery(service, row, binding) + committed_key = inspection["turn_key"] if inspection["committed_revalidation"] else None if committed_key is not None: service._bound(row, require_active=True) delegation_results.require_dependencies(service, binding, delegation_results.operation_brief(service, row)) @@ -188,27 +139,19 @@ def recover_validated_settlement( host_record=service._host_process_record(path)) record_revalidation_result(service, path, row, binding, result) return True - journal = validated_turn_journal(service, row, binding) - if journal is None: + if not inspection["validated_settlement"]: return False - decision = effect_runtime_result( - "collaboration.delegation.recover_validated_settlement", - { - "from": row["status"], - "identity_matched": True, - "journal_status": journal.get("status"), - "result_kind": journal.get("result_kind"), - "completed_phases": journal.get("completed_phases"), - "task_validation_passed": ( - isinstance(journal.get("task_validation"), dict) - and journal["task_validation"].get("ok") is True - ), - }, - ) + decision = effect_runtime_result("collaboration.delegation.recover_validated_settlement", { + "from": row["status"], "identity_matched": True, + "journal_status": inspection["status"], "result_kind": inspection["result_kind"], + "completed_phases": inspection["completed_phases"], + "task_validation_passed": inspection["task_validation_ok"], + }) + row["turn_key"] = inspection["turn_key"] row["status"] = decision["status"] row["turn_result"] = { - "status": journal.get("status"), - "result_kind": journal.get("result_kind"), + "status": inspection["status"], + "result_kind": inspection["result_kind"], "resume_turn_key": row["turn_key"], "reason": "validated Turn settlement requires same-operation recovery", "host_failure": None, diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index a8e037ef8f..48a8f329bd 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -266,6 +266,8 @@ export function createEffectRuntimeHandlers( ], ["turn_journal.write", lazyHandler(() => import("./turn_driver/turn_journal_effects.ts"), ({commitTurnJournal}) => commitTurnJournal)], ["turn_journal.find_settlement", lazyHandler(() => import("./turn_driver/turn_journal_query.ts"), ({findTurnJournalBySettlement}) => findTurnJournalBySettlement)], + ["collaboration.delegation.turn_recovery", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationTurnRecovery}) => delegationTurnRecovery)], + ["collaboration.delegation.recover_validated_settlement", lazyHandler(() => import("./collaboration/delegation.ts"), ({recoverValidatedDelegationSettlement}) => recoverValidatedDelegationSettlement)], ["turn_journal.observed_capabilities", lazyHandler(() => import("./turn_driver/turn_journal_query.ts"), ({readTurnJournalCapabilities}) => readTurnJournalCapabilities)], ["todo.completion_fence.evaluate", lazyHandler(() => import("./todos/completion_fence.ts"), ({evaluateTodoCompletionFence}) => evaluateTodoCompletionFence)], ["todo.field_update.plan", lazyHandler(() => import("./todos/field_update.ts"), ({planTodoFieldUpdate}) => planTodoFieldUpdate)], @@ -653,10 +655,8 @@ export function createEffectRuntimeHandlers( ["chat.turn.recovery_allowed", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({mayRecoverChatTurn}) => mayRecoverChatTurn)], ["collaboration.delegation.observe", lazyHandler(() => import("./collaboration/delegation.ts"), ({transitionDelegationObservation}) => transitionDelegationObservation)], ["collaboration.delegation.observe_wake", lazyHandler(() => import("./collaboration/delegation.ts"), ({decideDelegationWakeObservation}) => decideDelegationWakeObservation)], - ["collaboration.delegation.recover_validated_settlement", lazyHandler(() => import("./collaboration/delegation.ts"), ({recoverValidatedDelegationSettlement}) => recoverValidatedDelegationSettlement)], ["collaboration.delegation.stop", lazyHandler(() => import("./collaboration/delegation.ts"), ({decideDelegationStop}) => decideDelegationStop)], ["turn.task_validation_failure", lazyHandler(() => import("./turn_driver/task_validation_failure.ts"), ({taskValidationFailure}) => taskValidationFailure)], - ["collaboration.delegation.revalidated", lazyHandler(() => import("./collaboration/delegation.ts"), ({recoverRevalidatedDelegation}) => recoverRevalidatedDelegation)], ["collaboration.delegation.result_use_limits", lazyHandler(() => import("./collaboration/delegation_result_use.ts"), ({delegationResultUseLimits}) => delegationResultUseLimits)], ["collaboration.delegation.result_use", lazyHandler(() => import("./collaboration/delegation_result_use.ts"), ({qualifyDelegationResultUse}) => qualifyDelegationResultUse)], ["progress_review.evidence_scope", lazyHandler(() => import("./work_items/progress_review_evidence.ts"), ({progressReviewEvidenceScope}) => progressReviewEvidenceScope)], diff --git a/loopx/control_plane/turn_driver/turn_journal_query.ts b/loopx/control_plane/turn_driver/turn_journal_query.ts index 24e33e540a..62530abf9e 100644 --- a/loopx/control_plane/turn_driver/turn_journal_query.ts +++ b/loopx/control_plane/turn_driver/turn_journal_query.ts @@ -22,10 +22,12 @@ import { interpretTurnJournal, parseTurnJournalGoalBinding, } from "./turn_journal.ts"; +import { taskValidationFailure } from "./task_validation_failure.ts"; interface JournalEvidence { turn_key: string; observed_capabilities: string[] | null; + journal: JsonObject; } function decodeIdentity(value: unknown): BoundSettlementIdentity { @@ -122,11 +124,13 @@ function evidenceForJournal( if (!inspection.journal_consistent) { journalConflict(`Turn journal history is inconsistent: ${inspection.violations.join(", ")}`); } - return { + const evidence: JournalEvidence = { turn_key: turnKey, observed_capabilities: inspection.replay_legal ? observedCapabilities(jsonObject(plan?.turn_envelope) ?? {}) : null, + journal: journal!, }; + return evidence; } async function querySettlementJournal( @@ -193,6 +197,48 @@ export async function findTurnJournalBySettlement(params: JsonObject): Promise<{ return { turn_key: evidence?.turn_key ?? null }; } +/** One bounded read for delegation recovery; no second Python journal parse. */ +function delegationTaskFailure(journal: JsonObject): unknown { + return taskValidationFailure({ + result_kind: journal.result_kind, status: journal.status, + receipt: journal.receipt, validation: journal.task_validation, + ...(Object.hasOwn(journal, "validation_stage") ? {validation_stage: journal.validation_stage} : {}), + }).failure; +} + +function delegationRecoveryFacts(journal: JsonObject, turnKey: string): JsonObject { + const validation = jsonObject(journal.task_validation); + const hostResult = jsonObject(journal.host_result); + return { + status: journal.status ?? null, + result_kind: journal.result_kind ?? null, + task_failure: delegationTaskFailure(journal), + completed_phases: journal.completed_phases, + validated_settlement: journal.status === "in_progress" + && journal.result_kind === "validated_progress" + && JSON.stringify(journal.completed_phases) === JSON.stringify(["host_execute", "typed_result", "validation"]) + && validation?.ok === true && hostResult?.turn_key === turnKey + && hostResult?.result_kind === "validated_progress", + task_validation_ok: validation?.ok === true, + host_result_turn_key: hostResult?.turn_key === turnKey ? turnKey : null, + host_result_kind: hostResult?.result_kind === "validated_progress" ? "validated_progress" : null, + }; +} + +export async function findDelegationTurnJournal(params: JsonObject): Promise { + const identity = decodeIdentity(settlementIdentityPayload({ + goal_id: requireNonEmptyString(params.goal_id, "goal_id"), + agent_id: requireNonEmptyString(params.agent_id, "agent_id"), + todo_id: requireNonEmptyString(params.todo_id, "todo_id"), + turn_instance_id: requireNonEmptyString(params.turn_instance_id, "turn_instance_id"), + })); + const evidence = await querySettlementJournal(params.runtime_root, identity, null); + return { + turn_key: evidence?.turn_key ?? null, + recovery: evidence ? delegationRecoveryFacts(evidence.journal, evidence.turn_key) : null, + }; +} + export async function readTurnJournalCapabilities(params: JsonObject): Promise<{ observed_capabilities: string[] | null }> { const identity = decodeIdentity(params.settlement_identity); const goalRef = Object.hasOwn(params, "goal_ref") diff --git a/tests/control_plane_ts/turn_journal_query.test.ts b/tests/control_plane_ts/turn_journal_query.test.ts index ac2dd6bc28..120c6d54a4 100644 --- a/tests/control_plane_ts/turn_journal_query.test.ts +++ b/tests/control_plane_ts/turn_journal_query.test.ts @@ -6,6 +6,7 @@ import test, { type TestContext } from "node:test"; import { EffectRuntimeConflictError, EffectRuntimeRequestError } from "../../loopx/control_plane/effect_runtime_errors.ts"; import { commitTurnJournal } from "../../loopx/control_plane/turn_driver/turn_journal_effects.ts"; +import { delegationTurnRecovery } from "../../loopx/control_plane/collaboration/delegation.ts"; import { findTurnJournalBySettlement, readTurnJournalCapabilities } from "../../loopx/control_plane/turn_driver/turn_journal_query.ts"; const phases = ["host_execute", "typed_result", "validation", "durable_writeback", "quota_spend", "scheduler_apply", "scheduler_ack"]; @@ -77,6 +78,34 @@ test("real writer prefixes remain discoverable; only terminal history lends capa } }); +test("delegation recovery gets the validated journal decision from one typed read", async t => { + const { root, path } = await fixture(t); + const value = journal(); + Object.assign(value, { + status: "in_progress", + result_kind: "validated_progress", + completed_phases: ["host_execute", "typed_result", "validation"], + task_validation: {ok: true}, + host_result: {turn_key: key, result_kind: "validated_progress"}, + }); + await writeFile(path, JSON.stringify(value)); + + assert.deepEqual(await delegationTurnRecovery({ + runtime_root: root, ...selector, status: "rejected", + }), { + turn_key: key, + task_failure: null, + committed_revalidation: false, + validated_settlement: true, + completed_phases: ["host_execute", "typed_result", "validation"], + task_validation_ok: true, + task_validation: {ok: true}, + host_result: {turn_key: key, result_kind: "validated_progress"}, + status: "in_progress", + result_kind: "validated_progress", + }); +}); + test("capability evidence is bound to the current exact GoalRef", async t => { const { root, path } = await fixture(t); await writeFile(path, JSON.stringify(journal(goalRefA)));