diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4d8f913..d2f0b52 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,45 @@
 
 ## Unreleased
 
+- **Durable session-stack recovery (P0.1).** While a child session runs, the
+  parent's `state.json` records `active_child_session_id`; on `--resume` the
+  coordinator walks the pointer chain to the deepest live session instead of
+  silently reopening the parent and orphaning the running child. Terminal
+  children found at startup are finalized (children.json completed, pointer
+  cleared) and their parent resumed. Every interrupted-dispatch crash window
+  reconciles deterministically: dangling pointers are cleared, a fully
+  created child whose parent commit never landed is adopted, and leftover
+  request files never dispatch twice (children.json records the originating
+  `request_file`). Invalid child requests are terminally rejected
+  (`*.json.rejected`) instead of being re-read forever.
+- **Attempt ids.** Every dispatched task carries a unique `attempt_id`
+  (also on the wire in `TaskResponse`, echoed in `FinishedRequest`); a late
+  `/finished` from a superseded attempt of the same coordinates is treated as
+  stale rather than recorded as the current result.
+- Iteration artifacts (`result.json`, `result_text.txt`, `prompt.txt`,
+  `harness_run_id.txt`, `pending_finished_request.json`), `children.json`,
+  and `salvage.json` are all written atomically (unique temp + rename) — a
+  crash can never leave a truncated recovery artifact.
+- Internal: the three duplicated dispatch blocks in the coordinator collapsed
+  into one `_advance()` step (stop checks → child dispatch → next workflow →
+  stamped task), so they can no longer drift apart. `_advance()` also enforces
+  the suspended-parent invariant: a parent with a live child can never acquire
+  its own task (a duplicate `/finished` retry gets the child's live task
+  instead), and a coordinator-level transition lock serializes cross-store
+  handoffs.
+- Review hardening (adversarial Codex review of the above): request-file
+  tombstones apply only to RUNNING child records (a completed child's
+  filename is reusable for new work); the children.json record lands BEFORE
+  the child state so an interrupted dispatch is always discoverable
+  (`failed_dispatch` + exactly-once redispatch); startup reconciles every
+  running-projected record (terminal children finalize even without a
+  pointer); the first child task carries an attempt id; attempt checks are
+  strict whenever the live task has one (including `result.json` provenance —
+  a stale artifact can no longer complete a new attempt); semantically
+  unusable child requests (unknown workflow set, no eligible workflow) are
+  terminally rejected instead of wedging every completion; packaged prompts
+  instruct atomic control/goal_check publication; the crash model (process
+  crash, no fsync) is documented.
 - **The `pm_planner_dispatcher` template is executable from a clean init
   (P0.4).** `loopy init --template pm_planner_dispatcher` now also ships the
   `inner_outer_eval` child workflow set its dispatcher spawns — previously a
diff --git a/docs/http-contract.md b/docs/http-contract.md
index 2fbf966..55139af 100644
--- a/docs/http-contract.md
+++ b/docs/http-contract.md
@@ -33,6 +33,7 @@ Run response:
   "workflow_id": "planner",
   "session_id": "20260419_143022_71393ee22450_ab12cd34",
   "iteration": 3,
+  "attempt_id": "a1b2c3d4e5f6",
   "config_snapshot": {
     "goal": "Ship a minimal working landing page",
     "goal_hash": "71393ee22450",
@@ -115,6 +116,13 @@ Rules:
   escape hatch is to kill that process and register again — the 409 message
   names its pid.
 - If the loop is in a terminal state, `/register` immediately returns `action=stop`.
+- On `--resume`, the coordinator walks the durable parent→child session
+  pointers to the deepest live session: a running child continues where it
+  was (previously a restart silently reopened the parent and orphaned the
+  child), a child found terminal is finalized and its parent resumed, and
+  each interrupted-dispatch crash window reconciles deterministically
+  (dangling pointers cleared, fully-created-but-unpointed children adopted,
+  leftover request files never dispatched twice).
 
 ## POST /finished
 
@@ -128,6 +136,7 @@ Request:
   "success": true,
   "text": "done",
   "error": null,
+  "attempt_id": "a1b2c3d4e5f6",
   "worker": {
     "hostname": "buildbox",
     "pid": 4242,
@@ -143,7 +152,12 @@ Response: same shape as `/register` response (`action` is either `"run"` or `"st
 
 Rules:
 
-- If `session_id` + `workflow_id` + `iteration` does not match `current_task`,
+- Every dispatched task carries a unique `attempt_id`; the worker echoes it on
+  `/finished`. A call whose attempt id differs from the live task's — even
+  with matching session/workflow/iteration — is stale: it belongs to a
+  superseded attempt whose work was already recovered or abandoned.
+- If `session_id` + `workflow_id` + `iteration` (+ `attempt_id`, when both
+  sides carry one) does not match `current_task`,
   the call is treated as stale: state is not mutated and `current_task` is not
   changed. The current task's run response is returned only when the caller's
   identity matches the task's recorded owner (or either identity is unknown —
diff --git a/docs/session-layout.md b/docs/session-layout.md
index 90919ee..34f4398 100644
--- a/docs/session-layout.md
+++ b/docs/session-layout.md
@@ -165,6 +165,26 @@ eval-banana run \
 - If the file is missing but `result.json` exists for the active task, the
   coordinator can reconstruct the finished request from `result.json`
 
+`children.json` (parent sessions)
+
+- Index of child sessions dispatched by this session; each record carries the
+  child `session_id`, `workflow_set`, `status`, timestamps, `stop_reason`, and
+  the originating `request_file` name (which makes the dispatch scan
+  idempotent: a request whose filename already appears here is never
+  dispatched twice, even if a crash left the file behind)
+- The parent's `state.json` additionally records `active_child_session_id`
+  while a child runs — the durable session-stack pointer a restarted
+  coordinator follows to resume the child instead of orphaning it
+
+Atomicity and crash model: the coordinator- and worker-owned artifacts above
+are written atomically (same-directory temp file + rename), so a **process
+crash** never leaves them truncated — readers see either the previous or the
+new complete file. This does not extend to power loss / kernel crashes (no
+fsync), and it cannot be enforced for **workflow-written** signals
+(`control.json`, `goal_check.json`, child requests): the packaged prompts
+instruct agents to publish those via temp-file + rename, but a torn write by a
+non-compliant agent is read as invalid output.
+
 `salvage.json`
 
 - Written into the interrupted iteration's directory during crash recovery,
diff --git a/src/loopy_loop/coordinator_app.py b/src/loopy_loop/coordinator_app.py
index 81540bd..30098d5 100644
--- a/src/loopy_loop/coordinator_app.py
+++ b/src/loopy_loop/coordinator_app.py
@@ -5,8 +5,10 @@
 import logging
 from pathlib import Path
 import socket
+import threading
 from typing import Any
 from typing import TypeVar
+import uuid
 
 from fastapi import FastAPI
 from fastapi import HTTPException
@@ -47,6 +49,7 @@
 from loopy_loop.sessions import pending_finished_request_path
 from loopy_loop.sessions import result_path
 from loopy_loop.sessions import state_path
+from loopy_loop.sessions import write_json_atomic
 from loopy_loop.state_store import StateStore
 from loopy_loop.worker_identity import is_worker_alive
 
@@ -133,6 +136,14 @@ def __init__(
             preflight.workflow_set: preflight
         }
         self.state_store = state_store
+        # Serializes cross-store transitions (parent<->child handoff and the
+        # phase-B commits): FastAPI runs sync endpoints in a threadpool, so
+        # overlapping requests could otherwise race on self.state_store and
+        # the multi-file dispatch transition (C1). Reentrant because a child's
+        # terminal /finished resumes the parent by calling register_worker()
+        # on the same thread. Phase-A recovery (the long drain) deliberately
+        # runs OUTSIDE this lock on the initial register path.
+        self._transition_lock = threading.RLock()
         self._prepare_state(resume=resume)
 
     def register_worker(
@@ -146,7 +157,8 @@ def register_worker(
         # phase A when the state moved in between.
         for _ in range(3):
             recovery = self._plan_orphan_recovery()
-            response = self._register_attempt(caller=caller, recovery=recovery)
+            with self._transition_lock:
+                response = self._register_attempt(caller=caller, recovery=recovery)
             if response is not None:
                 return response
         raise WorkerBusyError(
@@ -235,50 +247,90 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse | None]:
                     # do not act on a stale plan — replan.
                     return current, None
 
-            # Step 4: Check stop conditions after abandoned-task cleanup.
-            stop_response = self._stop_response_if_needed(state=current)
-            if stop_response is not None:
-                return current, stop_response
-
-            child_response = self._dispatch_child_session_after_success(
-                state=current, caller=caller
-            )
-            if child_response is not None:
-                return current, child_response
-
-            # Step 5: Choose next workflow.
-            workflows = self._workflows_for(workflow_set=current.workflow_set)
-            workflow = choose_next_workflow(
-                workflows=workflows,
-                history=current.history,
-                iteration_count=current.iteration_count,
-            )
-            if workflow is None:
-                current.stop_reason = "no_eligible_workflow"
-                current.status = "failed"
-                return current, TaskResponse(
-                    action=STOP_ACTION, stop_reason="no_eligible_workflow"
-                )
-
-            # Step 6: Set current_task and return run response.
-            current.current_task = CurrentTask(
-                workflow_set=current.workflow_set,
-                workflow_id=workflow.id,
-                session_id=current.active_session_id,
-                iteration=current.iteration_count + 1,
-                started_at=now,
-                worker=caller,
-            )
-            return current, _build_run_response(
-                current_task=current.current_task,
-                config_snapshot=current.config_snapshot,
-            )
+            # Step 4+: stop conditions, child dispatch, next workflow.
+            return current, self._advance(state=current, caller=caller, now=now)
 
         response = self.state_store.mutate(mutator)
         for path in recovered_pending_paths:
             path.unlink(missing_ok=True)
         return response
 
+    def _suspended_parent_response(self, *, state: LoopState) -> TaskResponse | None:
+        """A parent with a live child must NEVER acquire its own current_task.
+
+        Reachable via overlapping /finished retries: the first call dispatches
+        the child and commits the suspended parent; a duplicate retry then
+        reads that parent with current_task=None and would otherwise advance
+        it — putting a parent task and a child task live simultaneously (the
+        C1 race from review). The duplicate gets the child's live task instead
+        (idempotent with the first response); a terminal child is finalized so
+        the advance continues as the legitimate parent resume.
+        """
+        child_id = state.active_child_session_id
+        if child_id is None:
+            return None
+        child_store = self._store_for(session_id=child_id)
+        child_state = child_store.read_state()
+        if child_state is None or child_store.is_terminal_state(state=child_state):
+            # Legitimate resume: finalize and let the advance continue.
+            if child_state is not None:
+                self._mark_child_record_complete(child_state=child_state)
+            state.active_child_session_id = None
+            return None
+        if child_state.current_task is not None:
+            return _build_run_response(
+                current_task=child_state.current_task,
+                config_snapshot=child_state.config_snapshot,
+            )
+        raise WorkerBusyError(
+            f"child session {child_id} is active; its next task is dispatched "
+            "through the child session, not the parent"
+        )
+
+    def _advance(
+        self, *, state: LoopState, caller: WorkerIdentity | None, now: datetime
+    ) -> TaskResponse:
+        """The single scheduling step shared by every dispatch path.
+
+        Order: stop conditions -> pending child dispatch -> next workflow ->
+        stamp a fresh CurrentTask (new attempt id, caller as owner). Extracted
+        so the three former copies (register, finished no-task, finished
+        matched) cannot drift apart.
+        """
+        suspended = self._suspended_parent_response(state=state)
+        if suspended is not None:
+            return suspended
+        stop_response = self._stop_response_if_needed(state=state)
+        if stop_response is not None:
+            return stop_response
+        child_response = self._dispatch_child_session_after_success(
+            state=state, caller=caller
+        )
+        if child_response is not None:
+            return child_response
+        workflows = self._workflows_for(workflow_set=state.workflow_set)
+        workflow = choose_next_workflow(
+            workflows=workflows,
+            history=state.history,
+            iteration_count=state.iteration_count,
+        )
+        if workflow is None:
+            state.stop_reason = "no_eligible_workflow"
+            state.status = "failed"
+            return TaskResponse(action=STOP_ACTION, stop_reason="no_eligible_workflow")
+        state.current_task = CurrentTask(
+            workflow_set=state.workflow_set,
+            workflow_id=workflow.id,
+            session_id=state.active_session_id,
+            iteration=state.iteration_count + 1,
+            started_at=now,
+            worker=caller,
+            attempt_id=_new_attempt_id(),
+        )
+        return _build_run_response(
+            current_task=state.current_task, config_snapshot=state.config_snapshot
+        )
+
     def _raise_if_worker_alive(self, *, current_task: CurrentTask) -> None:
         if is_worker_alive(current_task.worker) is not True:
             return
@@ -293,6 +345,12 @@ def _raise_if_worker_alive(self, *, current_task: CurrentTask) -> None:
 
     def finish_assignment(self, *, request: FinishedRequest) -> TaskResponse:
         caller = request.worker
+        with self._transition_lock:
+            return self._finish_assignment_locked(request=request, caller=caller)
+
+    def _finish_assignment_locked(
+        self, *, request: FinishedRequest, caller: WorkerIdentity | None
+    ) -> TaskResponse:
 
         def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]:
             current = _require_state(state=state)
@@ -301,39 +359,7 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]:
             # Step 3: No active task — stale call. Dispatch as if /register was called.
             # This handles the post-crash stale retry scenario safely.
             if current.current_task is None:
-                # Check stop conditions first; if terminal, return stop.
-                stop_response = self._stop_response_if_needed(state=current)
-                if stop_response is not None:
-                    return current, stop_response
-                child_response = self._dispatch_child_session_after_success(
-                    state=current, caller=caller
-                )
-                if child_response is not None:
-                    return current, child_response
-                workflows = self._workflows_for(workflow_set=current.workflow_set)
-                workflow = choose_next_workflow(
-                    workflows=workflows,
-                    history=current.history,
-                    iteration_count=current.iteration_count,
-                )
-                if workflow is None:
-                    current.stop_reason = "no_eligible_workflow"
-                    current.status = "failed"
-                    return current, TaskResponse(
-                        action=STOP_ACTION, stop_reason="no_eligible_workflow"
-                    )
-                current.current_task = CurrentTask(
-                    workflow_set=current.workflow_set,
-                    workflow_id=workflow.id,
-                    session_id=current.active_session_id,
-                    iteration=current.iteration_count + 1,
-                    started_at=now,
-                    worker=caller,
-                )
-                return current, _build_run_response(
-                    current_task=current.current_task,
-                    config_snapshot=current.config_snapshot,
-                )
+                return current, self._advance(state=current, caller=caller, now=now)
 
             # Step 4: Mismatch check — stale call for a different task.
             # Do NOT mutate state; return the current task's run response so the
@@ -345,6 +371,17 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]:
                 request.session_id != active.session_id
                 or request.workflow_id != active.workflow_id
                 or request.iteration != active.iteration
+                or (
+                    # A live task WITH an attempt id accepts only an exact
+                    # echo: a missing or different attempt means a superseded
+                    # or unversioned completion — its work was already
+                    # recovered/abandoned and redispatched. The wildcard
+                    # applies only when the persisted task itself predates
+                    # attempt ids (M5: legacy tolerance belongs to the OLD
+                    # task, never to an unversioned artifact vs a NEW task).
+                    active.attempt_id is not None
+                    and request.attempt_id != active.attempt_id
+                )
             ):
                 # Replaying the live task is only safe to its recorded owner:
                 # handing it to anyone else would start a second executor of
@@ -384,44 +421,8 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]:
                     action=STOP_ACTION, stop_reason="goal_check_broken"
                 )
 
-            # Step 7: Check stop conditions.
-            stop_response = self._stop_response_if_needed(state=current)
-            if stop_response is not None:
-                return current, stop_response
-
-            # Step 8: Dispatch next workflow.
-            child_response = self._dispatch_child_session_after_success(
-                state=current, caller=caller
-            )
-            if child_response is not None:
-                return current, child_response
-
-            workflows = self._workflows_for(workflow_set=current.workflow_set)
-            workflow = choose_next_workflow(
-                workflows=workflows,
-                history=current.history,
-                iteration_count=current.iteration_count,
-            )
-            if workflow is None:
-                current.stop_reason = "no_eligible_workflow"
-                current.status = "failed"
-                return current, TaskResponse(
-                    action=STOP_ACTION, stop_reason="no_eligible_workflow"
-                )
-
-            # Step 9: Set new current_task and return run response.
-            current.current_task = CurrentTask(
-                workflow_set=current.workflow_set,
-                workflow_id=workflow.id,
-                session_id=current.active_session_id,
-                iteration=current.iteration_count + 1,
-                started_at=now,
-                worker=caller,
-            )
-            return current, _build_run_response(
-                current_task=current.current_task,
-                config_snapshot=current.config_snapshot,
-            )
+            # Step 7+: stop conditions, child dispatch, next workflow.
+            return current, self._advance(state=current, caller=caller, now=now)
 
         response = self.state_store.mutate(mutator)
         if response.action == STOP_ACTION:
@@ -539,6 +540,14 @@ def _read_recoverable_finished_request(
         )
         if result is None:
             return None
+        if (
+            current_task.attempt_id is not None
+            and result.attempt_id != current_task.attempt_id
+        ):
+            # The artifact belongs to a superseded (or unversioned) attempt:
+            # accepting it would let a stale result complete the NEW attempt
+            # right after its stale pending file was correctly rejected (M5).
+            return None
         return (
             FinishedRequest(
                 session_id=current_task.session_id,
@@ -547,6 +556,7 @@ def _read_recoverable_finished_request(
                 success=result.success,
                 text=result.text,
                 error=result.error,
+                attempt_id=result.attempt_id,
             ),
             None,
         )
@@ -565,15 +575,159 @@ def _prepare_state(self, *, resume: bool) -> None:
                 "Found running loopy-loop state. Restart with --resume to continue "
                 "the in-progress session."
             )
+        active_state = self._reconstruct_session_stack(top_state=existing_state)
         create_session_dir(
             repo_root=self.repo_root,
-            session_id=existing_state.active_session_id,
-            goal_hash=existing_state.goal_hash,
-            goal=existing_state.config_snapshot.goal,
-            workflow_set=existing_state.workflow_set,
-            parent_session_id=existing_state.parent_session_id,
+            session_id=active_state.active_session_id,
+            goal_hash=active_state.goal_hash,
+            goal=active_state.config_snapshot.goal,
+            workflow_set=active_state.workflow_set,
+            parent_session_id=active_state.parent_session_id,
+        )
+
+    def _reconstruct_session_stack(self, *, top_state: LoopState) -> LoopState:
+        """Walk the durable parent->child pointers to the deepest live session.
+
+        A restarted coordinator previously reopened the latest TOP-LEVEL
+        session, silently orphaning a running child. Now:
+
+        - a non-terminal child pointed at by its parent becomes the active
+          session (the parent stays suspended, exactly as before the crash);
+        - a terminal child is finalized (children.json completed, pointer
+          cleared) and the walk resumes the parent;
+        - a dangling pointer (child state missing — the dispatch crashed
+          between commits) is cleared so the parent redispatches cleanly;
+        - a parent with NO pointer but a running children.json record whose
+          child is live ADOPTS it (the crash window where the child was fully
+          created but the parent commit never landed).
+        """
+        state = top_state
+        while True:
+            child_id = state.active_child_session_id or self._adoptable_child_id(
+                parent_state=state
+            )
+            if child_id is None:
+                return state
+            child_store = StateStore(
+                repo_root=self.repo_root,
+                state_path=state_path(repo_root=self.repo_root, session_id=child_id),
+            )
+            child_state = child_store.read_state()
+            parent_store = self._store_for(session_id=state.active_session_id)
+            if child_state is None:
+                logger.warning(
+                    "session %s points at child %s whose state is missing; "
+                    "clearing the pointer (interrupted dispatch)",
+                    state.active_session_id,
+                    child_id,
+                )
+                state = self._set_child_pointer(
+                    store=parent_store, child_session_id=None
+                )
+                return state
+            if child_store.is_terminal_state(state=child_state):
+                self._mark_child_record_complete(child_state=child_state)
+                self._clear_child_pointer(
+                    parent_store=parent_store,
+                    parent_session_id=state.active_session_id,
+                    child_session_id=child_id,
+                )
+                refreshed = parent_store.read_state()
+                if refreshed is None:  # pragma: no cover - state just existed
+                    raise RuntimeError("parent state vanished during recovery")
+                return refreshed
+            # Ensure the adopted case is persisted as a real pointer.
+            if state.active_child_session_id != child_id:
+                state = self._set_child_pointer(
+                    store=parent_store, child_session_id=child_id
+                )
+            self.state_store = child_store
+            state = child_state
+
+    def _adoptable_child_id(self, *, parent_state: LoopState) -> str | None:
+        """Reconcile EVERY running-projected child record, then return the
+        adoptable live child (if any).
+
+        Handles all the projections a crash (or a pre-pointer version of
+        loopy-loop) can leave behind:
+        - record running, child state TERMINAL -> finalize the record, remove
+          its leftover request file (previously ignored forever — M3);
+        - record running, child state MISSING -> mark the record
+          failed_dispatch so its request file redispatches exactly once (M2);
+        - record running, child state running and parent linkage correct ->
+          the crash window where the child was fully created but the parent's
+          pointer commit never landed: adopt it.
+        """
+        parent_session_id = parent_state.active_session_id
+        payload = _read_children_payload(
+            path=children_path(repo_root=self.repo_root, session_id=parent_session_id)
+        )
+        adoptable: str | None = None
+        changed = False
+        for record in payload["children"]:
+            if record.get("status") != "running":
+                continue
+            child_id = record.get("session_id")
+            if not child_id:
+                continue
+            child_store = self._store_for(session_id=child_id)
+            child_state = child_store.read_state()
+            if child_state is None:
+                record["status"] = "failed_dispatch"
+                record["stop_reason"] = "child state was never written"
+                changed = True
+                continue
+            if child_store.is_terminal_state(state=child_state):
+                self._mark_child_record_complete(child_state=child_state)
+                request_file = record.get("request_file")
+                if request_file:
+                    (
+                        child_requests_dir_path(
+                            repo_root=self.repo_root, session_id=parent_session_id
+                        )
+                        / request_file
+                    ).unlink(missing_ok=True)
+                continue
+            if child_state.parent_session_id != parent_session_id:
+                logger.warning(
+                    "child record %s does not link back to parent %s; ignoring",
+                    child_id,
+                    parent_session_id,
+                )
+                continue
+            if adoptable is not None:
+                logger.warning(
+                    "multiple running children recorded (%s, %s); adopting the "
+                    "newest and leaving the other for manual reconciliation",
+                    adoptable,
+                    child_id,
+                )
+            adoptable = child_id
+        if changed:
+            write_json_atomic(
+                path=children_path(
+                    repo_root=self.repo_root, session_id=parent_session_id
+                ),
+                payload=payload,
+            )
+        return adoptable
+
+    def _store_for(self, *, session_id: str) -> StateStore:
+        return StateStore(
+            repo_root=self.repo_root,
+            state_path=state_path(repo_root=self.repo_root, session_id=session_id),
         )
 
+    def _set_child_pointer(
+        self, *, store: StateStore, child_session_id: str | None
+    ) -> LoopState:
+        def mutator(state: LoopState | None) -> tuple[LoopState, LoopState]:
+            current = _require_state(state=state)
+            current.active_child_session_id = child_session_id
+            return current, current
+
+        return store.mutate(mutator)
+
     def _write_fresh_state(self) -> None:
         session_id = create_session_id(goal_hash=self.preflight.root_config.goal_hash)
         create_session_dir(
@@ -641,18 +795,40 @@ def _dispatch_child_session_if_requested(
         )
         if not requests_dir.exists():
             return None
+        dispatched_request_files = self._dispatched_request_files(
+            parent_session_id=state.active_session_id
+        )
         for request_path in sorted(requests_dir.glob("*.json")):
+            if request_path.name in dispatched_request_files:
+                # Already produced a child (the crash window between recording
+                # the child and unlinking the request): never dispatch twice.
+                request_path.unlink(missing_ok=True)
+                continue
             request = _read_signal(path=request_path, model=ChildSessionRequest)
             if request is None:
+                _reject_request(request_path, reason="invalid JSON or schema")
+                continue
+            # Total transition (M6): a schema-valid request that cannot be
+            # dispatched — unknown workflow set, broken workflow configs, or a
+            # set with no initially eligible workflow — must be terminally
+            # rejected, never left to wedge every future completion with the
+            # same error.
+            try:
+                preflight = self._preflight_for(
+                    workflow_set=request.workflow_set, goal=request.goal
+                )
+            except ConfigError as exc:
+                _reject_request(request_path, reason=str(exc))
                 continue
-            preflight = self._preflight_for(
-                workflow_set=request.workflow_set, goal=request.goal
-            )
             workflows = preflight.workflows
             workflow = choose_next_workflow(
                 workflows=workflows, history=[], iteration_count=0
             )
             if workflow is None:
+                _reject_request(
+                    request_path,
+                    reason="workflow set has no initially eligible workflow",
+                )
                 continue
             goal_hash = derive_goal_hash(goal=request.goal)
             child_session_id = create_session_id(goal_hash=goal_hash)
@@ -664,6 +840,22 @@ def _dispatch_child_session_if_requested(
                 workflow_set=request.workflow_set,
                 parent_session_id=state.active_session_id,
             )
+            # Durable intent FIRST (M2): the children.json record lands before
+            # the child state, so a crash in between leaves a discoverable
+            # "running record, missing state" projection that startup
+            # reconciliation marks failed_dispatch and redispatches — instead
+            # of an unindexed running-looking child no recovery can adopt.
+            self._append_child_record(
+                parent_session_id=state.active_session_id,
+                record=ChildSessionRecord(
+                    session_id=child_session_id,
+                    workflow_set=request.workflow_set,
+                    goal_hash=goal_hash,
+                    status="running",
+                    created_at=utc_now(),
+                    request_file=request_path.name,
+                ),
+            )
             snapshot = RootConfigSnapshot.model_validate(
                 preflight.root_config.model_dump(exclude=_COORDINATOR_ONLY_FIELDS)
             )
@@ -683,6 +875,7 @@ def _dispatch_child_session_if_requested(
                     iteration=1,
                     started_at=now,
                     worker=caller,
+                    attempt_id=_new_attempt_id(),
                 ),
             )
             child_store = StateStore(
@@ -695,16 +888,11 @@ def _dispatch_child_session_if_requested(
             child_task = child_state.current_task
             if child_task is None:
                 raise RuntimeError("Child session was created without a task")
-            self._append_child_record(
-                parent_session_id=state.active_session_id,
-                record=ChildSessionRecord(
-                    session_id=child_session_id,
-                    workflow_set=request.workflow_set,
-                    goal_hash=goal_hash,
-                    status="running",
-                    created_at=now,
-                ),
-            )
+            # The durable session-stack pointer: committed with the parent
+            # state when this mutator returns, so a restarted coordinator can
+            # walk parent -> child instead of resuming the parent and
+            # orphaning the running child.
+            state.active_child_session_id = child_session_id
             request_path.unlink(missing_ok=True)
             self.state_store = child_store
             return _build_run_response(
@@ -734,16 +922,72 @@ def _resume_parent_if_active_child_completed(
                 repo_root=self.repo_root, session_id=child_state.parent_session_id
             ),
         )
+        self._clear_child_pointer(
+            parent_store=parent_store,
+            parent_session_id=child_state.parent_session_id,
+            child_session_id=child_state.active_session_id,
+        )
         self.state_store = parent_store
         return self.register_worker(request=RegisterRequest(worker=caller))
 
+    def _clear_child_pointer(
+        self, *, parent_store: StateStore, parent_session_id: str, child_session_id: str
+    ) -> None:
+        """The child reached a terminal state: the parent's stack pointer no
+        longer points at live work. Also removes the originating request file
+        if a crash window left it behind (its children.json record already
+        prevents redispatch; this is just hygiene)."""
+
+        def mutator(state: LoopState | None) -> tuple[LoopState, None]:
+            parent = _require_state(state=state)
+            if parent.active_child_session_id == child_session_id:
+                parent.active_child_session_id = None
+            return parent, None
+
+        parent_store.mutate(mutator)
+        payload = _read_children_payload(
+            path=children_path(repo_root=self.repo_root, session_id=parent_session_id)
+        )
+        for record in payload["children"]:
+            if record.get("session_id") != child_session_id:
+                continue
+            request_file = record.get("request_file")
+            if request_file:
+                leftover = (
+                    child_requests_dir_path(
+                        repo_root=self.repo_root, session_id=parent_session_id
+                    )
+                    / request_file
+                )
+                leftover.unlink(missing_ok=True)
+            break
+
+    def _dispatched_request_files(self, *, parent_session_id: str) -> set[str]:
+        """Filenames suppressed by the crash-window tombstone.
+
+        Only RUNNING records suppress: a completed child's request filename is
+        legal to reuse for genuinely new work (a stable name like child.json
+        is a perfectly reasonable agent protocol). The crash window this
+        protects — record appended, request not yet unlinked — always has a
+        running record; completed leftovers are cleaned by the completion path
+        and by startup reconciliation.
+        """
+        path = children_path(repo_root=self.repo_root, session_id=parent_session_id)
+        payload = _read_children_payload(path=path)
+        return {
+            record["request_file"]
+            for record in payload["children"]
+            if record.get("request_file")
+            and record.get("status") in {"running", "dispatching"}
+        }
+
     def _append_child_record(
         self, *, parent_session_id: str, record: ChildSessionRecord
     ) -> None:
         path = children_path(repo_root=self.repo_root, session_id=parent_session_id)
         payload = _read_children_payload(path=path)
         payload["children"].append(json.loads(record.model_dump_json()))
-        path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+        write_json_atomic(path=path, payload=payload)
 
     def _mark_child_record_complete(self, *, child_state: LoopState) -> None:
         assert child_state.parent_session_id is not None
@@ -754,10 +998,15 @@ def _mark_child_record_complete(self, *, child_state: LoopState) -> None:
         for record in payload["children"]:
             if record.get("session_id") == child_state.active_session_id:
                 record["status"] = child_state.status
-                record["completed_at"] = utc_now().isoformat().replace("+00:00", "Z")
+                if not record.get("completed_at"):
+                    # Idempotent for audit: keep the FIRST observed completion
+                    # time across crash-replayed finalizations.
+                    record["completed_at"] = (
+                        utc_now().isoformat().replace("+00:00", "Z")
+                    )
                 record["stop_reason"] = child_state.stop_reason
                 break
-        path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+        write_json_atomic(path=path, payload=payload)
 
     def _apply_stop_precedence(self, *, state: LoopState) -> str | None:
         if state.goal_met:
@@ -828,6 +1077,10 @@ def _apply_session_control(self, *, state: LoopState) -> None:
         raise RuntimeError("unreachable")
 
 
+def _new_attempt_id() -> str:
+    return uuid.uuid4().hex[:12]
+
+
 def _build_run_response(
     *, current_task: CurrentTask, config_snapshot: RootConfigSnapshot
 ) -> TaskResponse:
@@ -837,6 +1090,7 @@ def _build_run_response(
         workflow_id=current_task.workflow_id,
         session_id=current_task.session_id,
         iteration=current_task.iteration,
+        attempt_id=current_task.attempt_id,
         config_snapshot=config_snapshot,
     )
 
@@ -847,11 +1101,34 @@ def _require_state(*, state: LoopState | None) -> LoopState:
     return state
 
 
+def _reject_request(request_path: Path, *, reason: str) -> None:
+    """Terminally reject a child request, keeping an inspectable record.
+
+    Collision-safe: a second rejection with the same original name never
+    overwrites the first record.
+    """
+    rejected = request_path.with_suffix(request_path.suffix + ".rejected")
+    if rejected.exists():
+        rejected = request_path.with_suffix(
+            request_path.suffix + f".{uuid.uuid4().hex[:8]}.rejected"
+        )
+    request_path.rename(rejected)
+    logger.warning(
+        "rejected child request %s (%s); kept as %s",
+        request_path.name,
+        reason,
+        rejected.name,
+    )
+
+
 def _same_task(a: CurrentTask, b: CurrentTask) -> bool:
     return (
         a.session_id == b.session_id
         and a.workflow_id == b.workflow_id
         and a.iteration == b.iteration
+        and (
+            a.attempt_id is None or b.attempt_id is None or a.attempt_id == b.attempt_id
+        )
     )
 
 
@@ -862,6 +1139,10 @@ def _matches_current_task(
         request.session_id == current_task.session_id
         and request.workflow_id == current_task.workflow_id
         and request.iteration == current_task.iteration
+        and (
+            current_task.attempt_id is None
+            or request.attempt_id == current_task.attempt_id
+        )
     )
 
 
diff --git a/src/loopy_loop/harness_runner.py b/src/loopy_loop/harness_runner.py
index ada3548..a874814 100644
--- a/src/loopy_loop/harness_runner.py
+++ b/src/loopy_loop/harness_runner.py
@@ -3,7 +3,6 @@
 import asyncio
 from collections.abc import Iterable
 import inspect
-import json
 from pathlib import Path
 import traceback
 from typing import Callable
@@ -23,6 +22,8 @@
 from loopy_loop.sessions import PROMPT_FILENAME
 from loopy_loop.sessions import RESULT_FILENAME
 from loopy_loop.sessions import RESULT_TEXT_FILENAME
+from loopy_loop.sessions import write_json_atomic
+from loopy_loop.sessions import write_text_atomic
 
 
 class TeamHarnessLike(Protocol):
@@ -166,17 +167,20 @@ def _failure_harness_paths(
 def write_iteration_artifacts(
     *, iteration_dir: Path, rendered_prompt: str, iteration_result: IterationResult
 ) -> None:
+    # Atomic writes throughout: result.json is what post-crash recovery trusts
+    # as proof of a completed iteration, and the rest should never exist
+    # truncated either.
     iteration_dir.mkdir(parents=True, exist_ok=True)
-    (iteration_dir / PROMPT_FILENAME).write_text(rendered_prompt, encoding="utf-8")
-    (iteration_dir / RESULT_TEXT_FILENAME).write_text(
-        iteration_result.text or "", encoding="utf-8"
+    write_text_atomic(path=iteration_dir / PROMPT_FILENAME, content=rendered_prompt)
+    write_text_atomic(
+        path=iteration_dir / RESULT_TEXT_FILENAME, content=iteration_result.text or ""
     )
-    (iteration_dir / HARNESS_RUN_ID_FILENAME).write_text(
-        iteration_result.harness_run_id, encoding="utf-8"
+    write_text_atomic(
+        path=iteration_dir / HARNESS_RUN_ID_FILENAME,
+        content=iteration_result.harness_run_id,
     )
-    payload = iteration_result.model_dump()
-    (iteration_dir / RESULT_FILENAME).write_text(
-        json.dumps(payload, indent=2), encoding="utf-8"
+    write_json_atomic(
+        path=iteration_dir / RESULT_FILENAME, payload=iteration_result.model_dump()
     )
 
 
diff --git a/src/loopy_loop/models.py b/src/loopy_loop/models.py
index 0fa0113..ef9442f 100644
--- a/src/loopy_loop/models.py
+++ b/src/loopy_loop/models.py
@@ -72,6 +72,10 @@ class CurrentTask(BaseModel):
     iteration: int = Field(...)
     started_at: datetime = Field(...)
     worker: WorkerIdentity | None = Field(default=None)
+    # Unique per dispatch: distinguishes a legitimate retry of
+    # (session, workflow, iteration) from a late /finished of an OLD attempt
+    # of the very same coordinates. None only on pre-attempt persisted state.
+    attempt_id: str | None = Field(default=None)
 
 
 class TaskResponse(BaseModel):
@@ -80,6 +84,7 @@ class TaskResponse(BaseModel):
     workflow_id: str | None = Field(default=None)
     session_id: str | None = Field(default=None)
     iteration: int | None = Field(default=None)
+    attempt_id: str | None = Field(default=None)
     config_snapshot: RootConfigSnapshot | None = Field(default=None)
     stop_reason: str | None = Field(default=None)
 
@@ -110,6 +115,11 @@ class LoopState(BaseModel):
     stop_reason: str | None = Field(default=None)
     iteration_count: int = Field(default=0)
     goal_check_consecutive_failures: int = Field(default=0)
+    # The durable session-stack pointer: while a child session is active, the
+    # parent records WHICH child, so a restarted coordinator can walk the
+    # chain to the deepest non-terminal session instead of silently resuming
+    # the parent and orphaning the running child.
+    active_child_session_id: str | None = Field(default=None)
     current_task: CurrentTask | None = Field(default=None)
     history: list[HistoryEntry] = Field(default_factory=list)
     config_snapshot: RootConfigSnapshot = Field(...)
@@ -125,6 +135,9 @@ class FinishedRequest(BaseModel):
     # Identity of the calling worker — the same worker will run the NEXT task
     # this response dispatches, so it is stamped onto that CurrentTask.
     worker: WorkerIdentity | None = Field(default=None)
+    # Echo of TaskResponse.attempt_id; lets the coordinator reject a late
+    # /finished from a superseded attempt of the same coordinates.
+    attempt_id: str | None = Field(default=None)
 
 
 class ControlSignal(BaseModel):
@@ -162,6 +175,9 @@ class IterationResult(BaseModel):
     error_detail: dict[str, object] | None = Field(default=None)
     harness_run_id: str = Field(default="")
     harness_output_dir: str = Field(default="")
+    # Attempt provenance: without it, a stale result.json could complete a
+    # NEW attempt right after its stale pending file was correctly rejected.
+    attempt_id: str | None = Field(default=None)
 
 
 class ChildSessionRequest(BaseModel):
@@ -185,3 +201,8 @@ class ChildSessionRecord(BaseModel):
     created_at: datetime = Field(...)
     completed_at: datetime | None = Field(default=None)
     stop_reason: str | None = Field(default=None)
+    # Name of the child_requests/ file that produced this child. Makes the
+    # dispatch scan idempotent across the crash window between recording the
+    # child and unlinking the request: a request whose filename already
+    # appears in children.json is never dispatched twice.
+    request_file: str | None = Field(default=None)
diff --git a/src/loopy_loop/recovery.py b/src/loopy_loop/recovery.py
index 4427135..85ae247 100644
--- a/src/loopy_loop/recovery.py
+++ b/src/loopy_loop/recovery.py
@@ -39,6 +39,7 @@
 from loopy_loop.models import utc_now
 from loopy_loop.sessions import iteration_dir_path
 from loopy_loop.sessions import iteration_harness_output_root
+from loopy_loop.sessions import write_json_atomic
 
 logger = logging.getLogger(__name__)
 
@@ -257,6 +258,4 @@ def _write_salvage_record(
         "unsettled_workers": outcome.unsettled_workers,
         "reports": outcome.reports,
     }
-    (iteration_dir / SALVAGE_FILENAME).write_text(
-        json.dumps(payload, indent=2), encoding="utf-8"
-    )
+    write_json_atomic(path=iteration_dir / SALVAGE_FILENAME, payload=payload)
diff --git a/src/loopy_loop/sessions.py b/src/loopy_loop/sessions.py
index 2dd4e7a..2372323 100644
--- a/src/loopy_loop/sessions.py
+++ b/src/loopy_loop/sessions.py
@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import json
+import os
 from pathlib import Path
+import tempfile
 import uuid
 
 from loopy_loop.config import LOOPY_DIRNAME
@@ -31,6 +33,32 @@
 GOAL_CHECK_FILENAME = "goal_check.json"
 
 
+def write_text_atomic(*, path: Path, content: str) -> None:
+    """Crash-safe file write: unique temp in the same directory + rename.
+
+    Recovery decisions are made from these artifacts, so a crash mid-write
+    must never leave a truncated file behind.
+    """
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, temp_name = tempfile.mkstemp(
+        dir=path.parent, prefix=path.name + ".", suffix=".tmp"
+    )
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(content)
+        os.replace(temp_name, path)
+    except BaseException:
+        try:
+            os.unlink(temp_name)
+        except OSError:
+            pass
+        raise
+
+
+def write_json_atomic(*, path: Path, payload: object) -> None:
+    write_text_atomic(path=path, content=json.dumps(payload, indent=2))
+
+
 def create_session_id(*, goal_hash: str) -> str:
     stamp = utc_now().strftime("%Y%m%d_%H%M%S")
     unique = uuid.uuid4().hex[:8]
diff --git a/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/eval_runner/prompt.txt b/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/eval_runner/prompt.txt
index 59197e6..a1a7aa8 100644
--- a/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/eval_runner/prompt.txt
+++ b/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/eval_runner/prompt.txt
@@ -86,3 +86,11 @@ Session control stop schema:
   "stop_reason": "goal_met",
   "schema_version": 1
 }
+
+Atomic file publication (important):
+- When updating the Session control path or writing goal_check.json, never
+  write the final path directly. Write a temporary file in the same directory
+  first, then rename it over the final path (e.g. `mv control.json.tmp
+  control.json`). The coordinator reads these files as state-machine inputs;
+  a truncated half-written file is treated as invalid output and can
+  terminate the session.
diff --git a/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/outer/prompt.txt b/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/outer/prompt.txt
index 4086d0e..d78778f 100644
--- a/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/outer/prompt.txt
+++ b/src/loopy_loop/templates/inner_outer_eval/.loopy_loop/workflow_sets/inner_outer_eval/workflows/outer/prompt.txt
@@ -334,3 +334,11 @@ You also have access agent-browser for any browser automation, web testing, scra
     - Before running commands, load the workflow guide once per session: `agent-browser skills get core` (or `--full` for the complete command reference). Specialized sub-skills: `electron`, `slack`, `dogfood`, `vercel-sandbox`, `agentcore` — load via `agent-browser skills get <name>`.
 
 Don't ask the human any questions.
+
+Atomic file publication (important):
+- When updating the Session control path or writing goal_check.json, never
+  write the final path directly. Write a temporary file in the same directory
+  first, then rename it over the final path (e.g. `mv control.json.tmp
+  control.json`). The coordinator reads these files as state-machine inputs;
+  a truncated half-written file is treated as invalid output and can
+  terminate the session.
diff --git a/src/loopy_loop/templates/pm_planner_dispatcher/.loopy_loop/workflow_sets/pm_planner_dispatcher/workflows/planner/prompt.txt b/src/loopy_loop/templates/pm_planner_dispatcher/.loopy_loop/workflow_sets/pm_planner_dispatcher/workflows/planner/prompt.txt
index e60d8d0..8c667af 100644
--- a/src/loopy_loop/templates/pm_planner_dispatcher/.loopy_loop/workflow_sets/pm_planner_dispatcher/workflows/planner/prompt.txt
+++ b/src/loopy_loop/templates/pm_planner_dispatcher/.loopy_loop/workflow_sets/pm_planner_dispatcher/workflows/planner/prompt.txt
@@ -97,3 +97,11 @@ Guardrails:
   useful.
 
 Use Codex.
+
+Atomic file publication (important):
+- When updating the Session control path or writing goal_check.json, never
+  write the final path directly. Write a temporary file in the same directory
+  first, then rename it over the final path (e.g. `mv control.json.tmp
+  control.json`). The coordinator reads these files as state-machine inputs;
+  a truncated half-written file is treated as invalid output and can
+  terminate the session.
diff --git a/src/loopy_loop/worker.py b/src/loopy_loop/worker.py
index e239eb7..d4a088f 100644
--- a/src/loopy_loop/worker.py
+++ b/src/loopy_loop/worker.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-import json
 from pathlib import Path
 import sys
 import time
@@ -33,6 +32,7 @@
 from loopy_loop.sessions import session_dir_path
 from loopy_loop.sessions import session_goal_path
 from loopy_loop.sessions import updates_from_user_path
+from loopy_loop.sessions import write_json_atomic
 from loopy_loop.worker_identity import current_worker_identity
 
 # Internal retry constants for /finished — not configurable externally.
@@ -200,6 +200,9 @@ def _run_task(
         iteration_result = IterationResult(
             success=False, text=None, error=str(exc), harness_run_id=""
         )
+    iteration_result = iteration_result.model_copy(
+        update={"attempt_id": task.attempt_id}
+    )
     write_iteration_artifacts(
         iteration_dir=iteration_dir,
         rendered_prompt=rendered_prompt,
@@ -213,6 +216,7 @@ def _run_task(
         text=iteration_result.text,
         error=iteration_result.error,
         worker=identity,
+        attempt_id=task.attempt_id,
     )
     pending_path = _write_pending_finished_request(
         repo_root=repo_root, request=finished_request
@@ -236,7 +240,9 @@ def _write_pending_finished_request(
         iteration=request.iteration,
         workflow_id=request.workflow_id,
     )
-    path.write_text(json.dumps(request.model_dump(), indent=2), encoding="utf-8")
+    # Crash-safe: this file is what post-crash recovery trusts as proof of a
+    # completed task — it must never exist truncated.
+    write_json_atomic(path=path, payload=request.model_dump())
     return path
 
 
diff --git a/src/tests/test_cli.py b/src/tests/test_cli.py
index b20455a..03963b8 100644
--- a/src/tests/test_cli.py
+++ b/src/tests/test_cli.py
@@ -309,6 +309,7 @@ def test_clean_pm_init_can_dispatch_an_inner_outer_eval_child(
             "success": True,
             "text": "planner selected an item",
             "worker": register_body["worker"],
+            "attempt_id": parent_task["attempt_id"],
         },
     ).json()
     assert child_task["action"] == "run"
diff --git a/src/tests/test_coordinator_app.py b/src/tests/test_coordinator_app.py
index e06613d..cfee6e2 100644
--- a/src/tests/test_coordinator_app.py
+++ b/src/tests/test_coordinator_app.py
@@ -76,6 +76,7 @@ def test_finished_records_history(repo_builder: Any, monkeypatch: Any) -> None:
             "workflow_id": reg["workflow_id"],
             "session_id": reg["session_id"],
             "iteration": reg["iteration"],
+            "attempt_id": reg.get("attempt_id"),
             "success": True,
             "text": "done",
             "error": None,
@@ -106,6 +107,7 @@ def test_finished_returns_next_run(repo_builder: Any, monkeypatch: Any) -> None:
             "workflow_id": reg["workflow_id"],
             "session_id": reg["session_id"],
             "iteration": reg["iteration"],
+            "attempt_id": reg.get("attempt_id"),
             "success": True,
             "text": "done",
             "error": None,
@@ -170,6 +172,7 @@ def test_child_session_runs_inside_parent_and_resumes_parent(
             "workflow_id": parent_task["workflow_id"],
             "session_id": parent_task["session_id"],
             "iteration": parent_task["iteration"],
+            "attempt_id": parent_task.get("attempt_id"),
             "success": True,
             "text": "parent planned child",
             "error": None,
@@ -207,6 +210,7 @@ def test_child_session_runs_inside_parent_and_resumes_parent(
             "workflow_id": child_task["workflow_id"],
             "session_id": child_task["session_id"],
             "iteration": child_task["iteration"],
+            "attempt_id": child_task.get("attempt_id"),
             "success": True,
             "text": "child done",
             "error": None,
@@ -276,6 +280,7 @@ def test_failed_parent_iteration_does_not_dispatch_child_request(
             "workflow_id": parent_task["workflow_id"],
             "session_id": parent_task["session_id"],
             "iteration": parent_task["iteration"],
+            "attempt_id": parent_task.get("attempt_id"),
             "success": False,
             "text": None,
             "error": "failed before child dispatch",
@@ -306,6 +311,7 @@ def test_finished_stale_mismatch_does_not_mutate(
             "workflow_id": reg["workflow_id"],
             "session_id": "wrong-session-id",
             "iteration": reg["iteration"],
+            "attempt_id": reg.get("attempt_id"),
             "success": True,
             "text": "done",
             "error": None,
@@ -605,6 +611,7 @@ def test_finished_stop_after_max_turns(repo_builder: Any, monkeypatch: Any) -> N
             "workflow_id": reg["workflow_id"],
             "session_id": reg["session_id"],
             "iteration": reg["iteration"],
+            "attempt_id": reg.get("attempt_id"),
             "success": True,
             "text": "done",
             "error": None,
diff --git a/src/tests/test_idempotent_finished.py b/src/tests/test_idempotent_finished.py
index 02701bc..5bdca5a 100644
--- a/src/tests/test_idempotent_finished.py
+++ b/src/tests/test_idempotent_finished.py
@@ -27,6 +27,7 @@ def test_stale_finished_mismatch_does_not_record_history_twice(
             "workflow_id": reg["workflow_id"],
             "session_id": reg["session_id"],
             "iteration": reg["iteration"],
+            "attempt_id": reg.get("attempt_id"),
             "success": True,
             "text": "done",
             "error": None,
@@ -42,6 +43,7 @@ def test_stale_finished_mismatch_does_not_record_history_twice(
             "workflow_id": reg["workflow_id"],
             "session_id": reg["session_id"],
             "iteration": reg["iteration"],
+            "attempt_id": reg.get("attempt_id"),
             "success": True,
             "text": "done again",
             "error": None,
@@ -77,6 +79,7 @@ def test_stale_finished_returns_current_task_run_response(
             "workflow_id": reg["workflow_id"],
             "session_id": "stale-session-id",
             "iteration": reg["iteration"],
+            "attempt_id": reg.get("attempt_id"),
             "success": True,
             "text": "done",
             "error": None,
diff --git a/src/tests/test_session_stack_recovery.py b/src/tests/test_session_stack_recovery.py
new file mode 100644
index 0000000..357d433
--- /dev/null
+++ b/src/tests/test_session_stack_recovery.py
@@ -0,0 +1,824 @@
+"""Tests for P0.1: durable session-stack / active-child crash recovery.
+
+A coordinator restart used to reopen the latest TOP-LEVEL session, silently
+orphaning a running child (its request file was already consumed, its state
+stayed "running" forever, and the parent could dispatch duplicate work). The
+parent now records a durable ``active_child_session_id`` pointer, startup
+walks the pointer chain to the deepest live session, and every crash window
+in the dispatch transition reconciles deterministically.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from fastapi.testclient import TestClient
+
+from loopy_loop.coordinator_app import create_coordinator_app
+from loopy_loop.recovery import RecoveryOutcome
+from loopy_loop.sessions import child_requests_dir_path
+from loopy_loop.sessions import children_path
+from loopy_loop.sessions import control_path
+from loopy_loop.sessions import state_path
+from loopy_loop.state_store import StateStore
+
+REGISTER_BODY = {"worker": {"hostname": "test-host", "pid": 999983, "starttime": None}}
+
+CHILD_WORKFLOW_CONFIG = "\n".join(
+    [
+        "enabled: true",
+        "run_every: 1",
+        "must_follow: null",
+        "not_before_iteration: 0",
+        "description: Child work",
+    ]
+)
+
+
+def _build_repo_with_child_set(repo_builder: Any) -> Any:
+    repo_root = repo_builder()
+    child_workflow_dir = (
+        repo_root
+        / ".loopy_loop"
+        / "workflow_sets"
+        / "child_set"
+        / "workflows"
+        / "child_work"
+    )
+    child_workflow_dir.mkdir(parents=True)
+    child_workflow_dir.joinpath("prompt.txt").write_text(
+        "Do the child work.", encoding="utf-8"
+    )
+    child_workflow_dir.joinpath("config.yaml").write_text(
+        CHILD_WORKFLOW_CONFIG + "\n", encoding="utf-8"
+    )
+    return repo_root
+
+
+def _dispatch_child(client: TestClient, repo_root: Any) -> tuple[dict, dict]:
+    """Drive the loop until a child session is dispatched; return (parent, child)."""
+    parent_task = client.post("/register", json=REGISTER_BODY).json()
+    request_dir = child_requests_dir_path(
+        repo_root=repo_root, session_id=parent_task["session_id"]
+    )
+    request_dir.mkdir(parents=True, exist_ok=True)
+    request_dir.joinpath("child.json").write_text(
+        json.dumps(
+            {
+                "workflow_set": "child_set",
+                "goal": "Handle a focused child task.",
+                "schema_version": 1,
+            }
+        ),
+        encoding="utf-8",
+    )
+    child_task = client.post(
+        "/finished",
+        json={
+            "workflow_id": parent_task["workflow_id"],
+            "session_id": parent_task["session_id"],
+            "iteration": parent_task["iteration"],
+            "success": True,
+            "text": "parent planned child",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": parent_task["attempt_id"],
+        },
+    ).json()
+    assert child_task["workflow_set"] == "child_set"
+    return parent_task, child_task
+
+
+def _parent_state(repo_root: Any, parent_session_id: str):
+    return StateStore(
+        repo_root=repo_root,
+        state_path=state_path(repo_root=repo_root, session_id=parent_session_id),
+    ).read_state()
+
+
+def _stub_recovery(monkeypatch: Any) -> None:
+    monkeypatch.setattr(
+        "loopy_loop.coordinator_app.recover_interrupted_iteration",
+        lambda **kwargs: RecoveryOutcome(),
+    )
+
+
+# ---------------------------------------------------------------------------
+# The durable pointer itself
+# ---------------------------------------------------------------------------
+
+
+def test_child_dispatch_records_parent_pointer_and_request_file(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+
+    parent = _parent_state(repo_root, parent_task["session_id"])
+    assert parent is not None
+    assert parent.active_child_session_id == child_task["session_id"]
+    records = json.loads(
+        children_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ).read_text(encoding="utf-8")
+    )
+    assert records["children"][0]["request_file"] == "child.json"
+
+
+def test_runtime_child_completion_clears_pointer(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+    control_path(repo_root=repo_root, session_id=child_task["session_id"]).write_text(
+        json.dumps(
+            {
+                "state": "stopped",
+                "reason": "child complete",
+                "stop_reason": "goal_met",
+                "schema_version": 1,
+            }
+        ),
+        encoding="utf-8",
+    )
+    resumed = client.post(
+        "/finished",
+        json={
+            "workflow_id": child_task["workflow_id"],
+            "session_id": child_task["session_id"],
+            "iteration": child_task["iteration"],
+            "success": True,
+            "text": "child done",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": child_task["attempt_id"],
+        },
+    ).json()
+    assert resumed["session_id"] == parent_task["session_id"]
+    parent = _parent_state(repo_root, parent_task["session_id"])
+    assert parent is not None
+    assert parent.active_child_session_id is None
+
+
+# ---------------------------------------------------------------------------
+# Coordinator restart scenarios (the crash gap this feature closes)
+# ---------------------------------------------------------------------------
+
+
+def test_restart_mid_child_resumes_the_child_not_the_parent(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+
+    # Coordinator "crashes": a brand-new app resumes from disk.
+    _stub_recovery(monkeypatch)
+    restarted = TestClient(create_coordinator_app(repo_root=repo_root, resume=True))
+    response = restarted.post("/register", json=REGISTER_BODY).json()
+
+    # The CHILD's work continues — previously the parent was silently resumed
+    # and the running child orphaned forever.
+    assert response["action"] == "run"
+    assert response["workflow_set"] == "child_set"
+    assert response["session_id"] == child_task["session_id"]
+    parent = _parent_state(repo_root, parent_task["session_id"])
+    assert parent is not None
+    assert parent.active_child_session_id == child_task["session_id"]
+
+
+def test_restart_after_child_terminal_finalizes_and_resumes_parent(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+
+    # The child reached a terminal state on disk, but the coordinator died
+    # before resuming the parent (the exact window Codex's earlier review
+    # flagged as untested).
+    child_store = StateStore(
+        repo_root=repo_root,
+        state_path=state_path(repo_root=repo_root, session_id=child_task["session_id"]),
+    )
+    child = child_store.read_state()
+    assert child is not None
+    child.status = "goal_met"
+    child.stop_reason = "goal_met"
+    child.current_task = None
+    child_store.write_state(state=child)
+
+    restarted = TestClient(create_coordinator_app(repo_root=repo_root, resume=True))
+    response = restarted.post("/register", json=REGISTER_BODY).json()
+
+    assert response["action"] == "run"
+    assert response["session_id"] == parent_task["session_id"]
+    parent = _parent_state(repo_root, parent_task["session_id"])
+    assert parent is not None
+    assert parent.active_child_session_id is None
+    records = json.loads(
+        children_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ).read_text(encoding="utf-8")
+    )
+    assert records["children"][0]["status"] == "goal_met"
+
+
+def test_restart_with_dangling_pointer_clears_it(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # The dispatch crashed between commits: the pointer exists but the child
+    # state was never written. The parent must recover cleanly.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task = client.post("/register", json=REGISTER_BODY).json()
+    store = StateStore(repo_root=repo_root)
+    state = store.read_state()
+    assert state is not None
+    state.active_child_session_id = "20990101_000000_deadbeef_missing0"
+    store.write_state(state=state)
+
+    restarted = TestClient(create_coordinator_app(repo_root=repo_root, resume=True))
+    _stub_recovery(monkeypatch)
+    response = restarted.post("/register", json=REGISTER_BODY).json()
+    assert response["action"] == "run"
+    assert response["session_id"] == parent_task["session_id"]
+    parent = _parent_state(repo_root, parent_task["session_id"])
+    assert parent is not None
+    assert parent.active_child_session_id is None
+
+
+def test_restart_adopts_running_child_when_pointer_never_committed(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # Crash window: child fully created + children.json recorded, but the
+    # parent state commit (which carries the pointer) never landed.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+    parent_store = StateStore(
+        repo_root=repo_root,
+        state_path=state_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ),
+    )
+    parent = parent_store.read_state()
+    assert parent is not None
+    parent.active_child_session_id = None  # simulate the lost commit
+    parent_store.write_state(state=parent)
+
+    _stub_recovery(monkeypatch)
+    restarted = TestClient(create_coordinator_app(repo_root=repo_root, resume=True))
+    response = restarted.post("/register", json=REGISTER_BODY).json()
+
+    assert response["session_id"] == child_task["session_id"]
+    parent = _parent_state(repo_root, parent_task["session_id"])
+    assert parent is not None
+    assert parent.active_child_session_id == child_task["session_id"]  # re-adopted
+
+
+# ---------------------------------------------------------------------------
+# Request-file idempotency and rejection
+# ---------------------------------------------------------------------------
+
+
+def test_leftover_request_file_never_dispatches_twice(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # Crash window: children.json recorded the child but the request unlink
+    # never happened. After the child completes, the leftover file must not
+    # spawn a duplicate child.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+
+    request_dir = child_requests_dir_path(
+        repo_root=repo_root, session_id=parent_task["session_id"]
+    )
+    request_dir.joinpath("child.json").write_text(  # resurrect the consumed file
+        json.dumps(
+            {
+                "workflow_set": "child_set",
+                "goal": "Handle a focused child task.",
+                "schema_version": 1,
+            }
+        ),
+        encoding="utf-8",
+    )
+    control_path(repo_root=repo_root, session_id=child_task["session_id"]).write_text(
+        json.dumps(
+            {
+                "state": "stopped",
+                "reason": "child complete",
+                "stop_reason": "goal_met",
+                "schema_version": 1,
+            }
+        ),
+        encoding="utf-8",
+    )
+    resumed = client.post(
+        "/finished",
+        json={
+            "workflow_id": child_task["workflow_id"],
+            "session_id": child_task["session_id"],
+            "iteration": child_task["iteration"],
+            "success": True,
+            "text": "child done",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": child_task["attempt_id"],
+        },
+    ).json()
+    assert resumed["session_id"] == parent_task["session_id"]
+    records = json.loads(
+        children_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ).read_text(encoding="utf-8")
+    )
+    assert len(records["children"]) == 1  # no duplicate child
+    assert not request_dir.joinpath("child.json").exists()  # hygiene
+
+
+def test_invalid_child_request_is_rejected_terminally(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task = client.post("/register", json=REGISTER_BODY).json()
+    request_dir = child_requests_dir_path(
+        repo_root=repo_root, session_id=parent_task["session_id"]
+    )
+    request_dir.mkdir(parents=True, exist_ok=True)
+    request_dir.joinpath("broken.json").write_text("{not json", encoding="utf-8")
+
+    response = client.post(
+        "/finished",
+        json={
+            "workflow_id": parent_task["workflow_id"],
+            "session_id": parent_task["session_id"],
+            "iteration": parent_task["iteration"],
+            "success": True,
+            "text": "done",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": parent_task["attempt_id"],
+        },
+    ).json()
+    assert response["action"] == "run"
+    assert response["workflow_set"] == "main"  # normal dispatch, no child
+    assert not request_dir.joinpath("broken.json").exists()
+    assert request_dir.joinpath("broken.json.rejected").exists()
+
+
+# ---------------------------------------------------------------------------
+# Attempt ids
+# ---------------------------------------------------------------------------
+
+
+def test_dispatch_carries_attempt_id_and_stale_attempt_is_not_processed(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = repo_builder()
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    task = client.post("/register", json=REGISTER_BODY).json()
+    assert task["attempt_id"], "every dispatch must carry an attempt id"
+
+    # Same coordinates, WRONG attempt: a late /finished from a superseded
+    # attempt must be treated as stale (owner gets the live-task replay,
+    # state is not mutated), never processed as the current result.
+    stale = client.post(
+        "/finished",
+        json={
+            "workflow_id": task["workflow_id"],
+            "session_id": task["session_id"],
+            "iteration": task["iteration"],
+            "success": True,
+            "text": "late result from a previous attempt",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": "superseded0000",
+        },
+    ).json()
+    assert stale["action"] == "run"
+    assert stale["attempt_id"] == task["attempt_id"]  # the live attempt
+    state = StateStore(repo_root=repo_root).read_state()
+    assert state is not None
+    assert state.history == []  # nothing was recorded
+
+    # Correct attempt: processed normally.
+    done = client.post(
+        "/finished",
+        json={
+            "workflow_id": task["workflow_id"],
+            "session_id": task["session_id"],
+            "iteration": task["iteration"],
+            "success": True,
+            "text": "done",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": task["attempt_id"],
+        },
+    ).json()
+    assert done["action"] == "run"
+    state = StateStore(repo_root=repo_root).read_state()
+    assert state is not None
+    assert len(state.history) == 1
+
+
+# ---------------------------------------------------------------------------
+# Review-driven coverage (Codex P0.1 review: C1, M1-M6, m1)
+# ---------------------------------------------------------------------------
+
+
+def test_duplicate_finished_never_gives_suspended_parent_a_task(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # C1: overlapping /finished retries. The first dispatches the child and
+    # commits the suspended parent; a duplicate retry must NOT advance the
+    # parent (parent task + child task live simultaneously). It gets the
+    # child's live task instead — idempotent with the first response.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+
+    duplicate = client.post(
+        "/finished",
+        json={
+            "workflow_id": parent_task["workflow_id"],
+            "session_id": parent_task["session_id"],
+            "iteration": parent_task["iteration"],
+            "success": True,
+            "text": "parent planned child",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": parent_task["attempt_id"],
+        },
+    ).json()
+    assert duplicate["action"] == "run"
+    assert duplicate["session_id"] == child_task["session_id"]  # child, not parent
+    parent = _parent_state(repo_root, parent_task["session_id"])
+    assert parent is not None
+    assert parent.current_task is None  # the invariant C1 violated
+    assert parent.active_child_session_id == child_task["session_id"]
+
+
+def test_completed_childs_request_filename_is_reusable(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # M1: the tombstone must apply only while the record is running — a later,
+    # genuinely new request under the same filename must dispatch.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    # Two independent parent workflows: the default repo's goal_check would
+    # be dispatched after the resume and fail (no goal_check.json), blocking
+    # the after-success child scan this test needs; a single workflow would
+    # starve the resume (run_every not yet satisfied).
+    repo_root = repo_builder(
+        workflows={
+            "planner": {
+                "prompt": "Plan work.",
+                "config": {
+                    "enabled": True,
+                    "run_every": 1,
+                    "must_follow": None,
+                    "not_before_iteration": 0,
+                    "description": "Plan",
+                },
+            },
+            "implement": {
+                "prompt": "Implement.",
+                "config": {
+                    "enabled": True,
+                    "run_every": 1,
+                    "must_follow": None,
+                    "not_before_iteration": 0,
+                    "description": "Implement",
+                },
+            },
+        }
+    )
+    child_workflow_dir = (
+        repo_root
+        / ".loopy_loop"
+        / "workflow_sets"
+        / "child_set"
+        / "workflows"
+        / "child_work"
+    )
+    child_workflow_dir.mkdir(parents=True)
+    child_workflow_dir.joinpath("prompt.txt").write_text(
+        "Do the child work.", encoding="utf-8"
+    )
+    child_workflow_dir.joinpath("config.yaml").write_text(
+        CHILD_WORKFLOW_CONFIG + "\n", encoding="utf-8"
+    )
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+    control_path(repo_root=repo_root, session_id=child_task["session_id"]).write_text(
+        json.dumps(
+            {
+                "state": "stopped",
+                "reason": "done",
+                "stop_reason": "goal_met",
+                "schema_version": 1,
+            }
+        ),
+        encoding="utf-8",
+    )
+    resumed_parent = client.post(
+        "/finished",
+        json={
+            "workflow_id": child_task["workflow_id"],
+            "session_id": child_task["session_id"],
+            "iteration": child_task["iteration"],
+            "success": True,
+            "text": "child A done",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": child_task["attempt_id"],
+        },
+    ).json()
+    assert resumed_parent["session_id"] == parent_task["session_id"]
+
+    # A NEW request reusing the same filename for new work:
+    request_dir = child_requests_dir_path(
+        repo_root=repo_root, session_id=parent_task["session_id"]
+    )
+    request_dir.joinpath("child.json").write_text(
+        json.dumps(
+            {
+                "workflow_set": "child_set",
+                "goal": "Second work item.",
+                "schema_version": 1,
+            }
+        ),
+        encoding="utf-8",
+    )
+    second_child = client.post(
+        "/finished",
+        json={
+            "workflow_id": resumed_parent["workflow_id"],
+            "session_id": resumed_parent["session_id"],
+            "iteration": resumed_parent["iteration"],
+            "success": True,
+            "text": "planned item B",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": resumed_parent["attempt_id"],
+        },
+    ).json()
+    assert second_child["workflow_set"] == "child_set"
+    records = json.loads(
+        children_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ).read_text(encoding="utf-8")
+    )
+    assert len(records["children"]) == 2  # B was dispatched, not swallowed
+
+
+def test_restart_reconciles_record_without_child_state(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # M2 window (record lands before child state): a running record whose
+    # child state is missing is marked failed_dispatch and its request file
+    # redispatches exactly once.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+
+    # Simulate the crash: erase the child's state (record + request remain).
+    child_state_path = state_path(
+        repo_root=repo_root, session_id=child_task["session_id"]
+    )
+    child_state_path.unlink()
+    parent_store = StateStore(
+        repo_root=repo_root,
+        state_path=state_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ),
+    )
+    parent = parent_store.read_state()
+    assert parent is not None
+    parent.active_child_session_id = None  # pointer commit never landed
+    parent_store.write_state(state=parent)
+    request_dir = child_requests_dir_path(
+        repo_root=repo_root, session_id=parent_task["session_id"]
+    )
+    request_dir.joinpath("child.json").write_text(  # unlink never happened
+        json.dumps(
+            {
+                "workflow_set": "child_set",
+                "goal": "Handle a focused child task.",
+                "schema_version": 1,
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    _stub_recovery(monkeypatch)
+    restarted = TestClient(create_coordinator_app(repo_root=repo_root, resume=True))
+    response = restarted.post("/register", json=REGISTER_BODY).json()
+    records = json.loads(
+        children_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ).read_text(encoding="utf-8")
+    )
+    statuses = [record["status"] for record in records["children"]]
+    assert "failed_dispatch" in statuses  # the broken dispatch is closed out
+    # The request redispatches (either already in this response or on the
+    # next successful parent iteration) — the file must not be tombstoned.
+    assert response["action"] == "run"
+
+
+def test_restart_finalizes_terminal_child_without_pointer(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # M3: terminal child + running record + NO pointer (pre-pointer version
+    # crash) was ignored forever. Reconciliation must finalize it.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+
+    child_store = StateStore(
+        repo_root=repo_root,
+        state_path=state_path(repo_root=repo_root, session_id=child_task["session_id"]),
+    )
+    child = child_store.read_state()
+    assert child is not None
+    child.status = "goal_met"
+    child.stop_reason = "goal_met"
+    child.current_task = None
+    child_store.write_state(state=child)
+    parent_store = StateStore(
+        repo_root=repo_root,
+        state_path=state_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ),
+    )
+    parent = parent_store.read_state()
+    assert parent is not None
+    parent.active_child_session_id = None  # pre-pointer state
+    parent_store.write_state(state=parent)
+
+    restarted = TestClient(create_coordinator_app(repo_root=repo_root, resume=True))
+    response = restarted.post("/register", json=REGISTER_BODY).json()
+    assert response["session_id"] == parent_task["session_id"]
+    records = json.loads(
+        children_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ).read_text(encoding="utf-8")
+    )
+    assert records["children"][0]["status"] == "goal_met"  # finalized, not stuck
+
+
+def test_first_child_task_carries_an_attempt_id(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # M4: the initial child dispatch bypassed _advance and shipped a null
+    # attempt, leaving the child's first iteration unfenced.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    _, child_task = _dispatch_child(client, repo_root)
+    assert child_task["attempt_id"]
+
+
+def test_stale_result_artifact_cannot_complete_a_new_attempt(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # M5 case 3: a stale pending file is rejected, but the stale result.json
+    # right next to it must not slip through as the NEW attempt's completion.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = repo_builder()
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    task = client.post("/register", json=REGISTER_BODY).json()
+
+    from loopy_loop.sessions import ensure_iteration_dir
+
+    iteration_dir = ensure_iteration_dir(
+        repo_root=repo_root,
+        session_id=task["session_id"],
+        iteration=task["iteration"],
+        workflow_id=task["workflow_id"],
+    )
+    iteration_dir.joinpath("result.json").write_text(
+        json.dumps(
+            {
+                "success": True,
+                "text": "stale result from a superseded attempt",
+                "error": None,
+                "harness_run_id": "old",
+                "harness_output_dir": "",
+                "attempt_id": "superseded0000",
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(
+        "loopy_loop.coordinator_app.is_worker_alive", lambda identity: False
+    )
+    _stub_recovery(monkeypatch)
+    response = client.post("/register", json=REGISTER_BODY).json()
+    assert response["action"] == "run"
+    state = StateStore(repo_root=repo_root).read_state()
+    assert state is not None
+    # The stale artifact was NOT recorded as a successful completion:
+    assert state.history[0].success is False
+    assert state.history[0].error in {"abandoned", "abandoned_after_drain"}
+
+
+def test_semantically_invalid_child_request_is_rejected_not_wedging(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # M6: a schema-valid request naming an unknown workflow set previously
+    # raised out of the mutator — HTTP 500 on EVERY completion, forever.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = repo_builder()
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task = client.post("/register", json=REGISTER_BODY).json()
+    request_dir = child_requests_dir_path(
+        repo_root=repo_root, session_id=parent_task["session_id"]
+    )
+    request_dir.mkdir(parents=True, exist_ok=True)
+    request_dir.joinpath("bad.json").write_text(
+        json.dumps(
+            {"workflow_set": "does_not_exist", "goal": "x", "schema_version": 1}
+        ),
+        encoding="utf-8",
+    )
+    response = client.post(
+        "/finished",
+        json={
+            "workflow_id": parent_task["workflow_id"],
+            "session_id": parent_task["session_id"],
+            "iteration": parent_task["iteration"],
+            "success": True,
+            "text": "done",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": parent_task["attempt_id"],
+        },
+    )
+    assert response.status_code == 200  # the completion is committed
+    assert response.json()["action"] == "run"
+    assert not request_dir.joinpath("bad.json").exists()
+    rejected = list(request_dir.glob("bad.json*.rejected")) + list(
+        request_dir.glob("bad.json.rejected")
+    )
+    assert rejected, "the unusable request must be terminally rejected"
+
+
+def test_double_finalization_keeps_first_completed_at(
+    repo_builder: Any, monkeypatch: Any
+) -> None:
+    # m1: crash-replayed finalization must not rewrite the audit timestamp.
+    monkeypatch.setenv("OPENROUTER_API_KEY", "secret")
+    repo_root = _build_repo_with_child_set(repo_builder)
+    client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False))
+    parent_task, child_task = _dispatch_child(client, repo_root)
+    control_path(repo_root=repo_root, session_id=child_task["session_id"]).write_text(
+        json.dumps(
+            {
+                "state": "stopped",
+                "reason": "done",
+                "stop_reason": "goal_met",
+                "schema_version": 1,
+            }
+        ),
+        encoding="utf-8",
+    )
+    client.post(
+        "/finished",
+        json={
+            "workflow_id": child_task["workflow_id"],
+            "session_id": child_task["session_id"],
+            "iteration": child_task["iteration"],
+            "success": True,
+            "text": "child done",
+            "worker": REGISTER_BODY["worker"],
+            "attempt_id": child_task["attempt_id"],
+        },
+    )
+    children_file = children_path(
+        repo_root=repo_root, session_id=parent_task["session_id"]
+    )
+    first = json.loads(children_file.read_text())["children"][0]["completed_at"]
+
+    # Simulate the crash-replay: re-finalize at startup (pointer re-set).
+    parent_store = StateStore(
+        repo_root=repo_root,
+        state_path=state_path(
+            repo_root=repo_root, session_id=parent_task["session_id"]
+        ),
+    )
+    parent = parent_store.read_state()
+    assert parent is not None
+    parent.active_child_session_id = child_task["session_id"]
+    parent_store.write_state(state=parent)
+    TestClient(create_coordinator_app(repo_root=repo_root, resume=True))
+    second = json.loads(children_file.read_text())["children"][0]["completed_at"]
+    assert second == first
diff --git a/website/src/app/docs/http-contract/page.mdx b/website/src/app/docs/http-contract/page.mdx
index 52db419..d55cef4 100644
--- a/website/src/app/docs/http-contract/page.mdx
+++ b/website/src/app/docs/http-contract/page.mdx
@@ -181,6 +181,18 @@ acknowledgment never corrupts loop state.
 coordinator does **not** mutate state and does **not** change the current task; it
 returns the current task's run response so the caller sees what is actually running.
 
+**Attempt ids.** Every dispatched task carries a unique `attempt_id`, echoed by
+the worker on `/finished`. A call from a superseded attempt — same
+session/workflow/iteration, different attempt — is treated as stale, so a late
+result from work that was already recovered or abandoned can never be recorded
+as the current iteration's outcome.
+
+**Session-stack recovery on `--resume`.** The coordinator walks the durable
+parent→child pointers to the deepest live session: a running child continues
+where it was (previously a restart silently reopened the parent and orphaned
+the child), a terminal child is finalized and its parent resumed, and every
+interrupted-dispatch crash window reconciles deterministically.
+
 **Finish with no active task.** If `/finished` arrives when no task is current, the
 coordinator dispatches the next available task as if `/register` had been called —
 or returns `action: "stop"` if the state is terminal. This makes a post-crash retry
diff --git a/website/src/app/docs/session-layout/page.mdx b/website/src/app/docs/session-layout/page.mdx
index a838df5..9d1654d 100644
--- a/website/src/app/docs/session-layout/page.mdx
+++ b/website/src/app/docs/session-layout/page.mdx
@@ -175,6 +175,15 @@ task instead of marking it `abandoned`; if it is missing but `result.json` exist
 for the active task, the coordinator reconstructs the finished request from
 `result.json`. See [HTTP Contract](/docs/http-contract) for the recovery rules.
 
+**`children.json`** (parent sessions) — Index of child sessions dispatched by
+this session. Each record carries the child `session_id`, `workflow_set`,
+`status`, timestamps, `stop_reason`, and the originating `request_file` name,
+which makes the dispatch scan idempotent — a request whose filename already
+appears here is never dispatched twice, even if a crash left the file behind.
+While a child runs, the parent's `state.json` records
+`active_child_session_id`: the durable session-stack pointer a restarted
+coordinator follows to resume the child instead of orphaning it.
+
 **`salvage.json`** — Written into the interrupted iteration's directory during
 crash recovery, when the coordinator drained or reaped agent processes a dead
 worker's harness run left behind (per `recovery_policy`). It records the reap