refactor(delegation): collapse turn recovery bridge into TypeScript

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:

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<JsonObject> {
+  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. */

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,

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)],