fix(runner): stamp event generations and record an earlier generation's events as history (#19960)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs execute through the Paperclip Runner, which turns provider output (Codex, OpenCode, Claude through ACPX) into durable runner events > - In the restart chaos campaign, events from an earlier turn or runner generation failed the current turn about ten times (RC-15, 20, 20b, 20c, 33, 33b, 59 and 61) > - #19718, #19722, #19772 and #19783 each special-cased one variant, and other variants still fail the run > - The durable-ownership plan (section G, invariant 6) asks for one rule: every event carries its generation and turn, and an earlier generation's events are recorded as history and never fail the current turn > - This pull request stamps the runner generation on every event and applies one stale-event rule in the ACPX provider state, the Codex driver (which Runnerd also uses for OpenCode) and the pending `turn.start` check > - The benefit is that a late report about a settled or restored turn no longer stops a healthy turn, while events for a turn the runner never saw still fail closed ## Linked Issues or Issue Description Refs #19718, #19722, #19772, #19783 **What happened?** A runner or provider can report a turn after that turn settled: a resumed provider reports the lost turn's usage or terminal, an ACPX sidecar emits a last delta after the turn terminal, or a held failure of the previous turn sits in front of a new `turn.start`. Several of these paths fail the current turn: ACPX `AcpxEventScope` rejects any event for a non-active turn ("named a stale turn"), the Codex driver fails `turn_binding_mismatch` for a settled turn's late item between turns, and `runnerdPendingTurnStartFailure` reports any held non-recoverable failure, whatever turn it names. Runner events carried no generation. **Expected behavior** An event for a turn that is not active, and that an earlier generation accepted or this generation already settled, is recorded as history and never fails the current turn. An event for a turn the runner never saw still fails closed. **Steps to reproduce** ACPX: settle turn 1, start turn 2, and deliver a late `runtime.event` for turn 1. Before this change, the provider state returns an error and the run fails. ## What Changed - Runner (`durable/state.rs`, `durable/runner.rs`): every event payload carries `generation`, the runner process generation that emitted it. A runner process that recovers state starts the next generation (`begin_recovered_generation`); it is saved with the next normal write, so recovery adds no write before the first turn. Loading stays pure. Old saved state loads as generation 0. Outbox events keep their original generation. - PRP event schemas v1–v3 declare the optional `generation` property; generated validators and the manifest are regenerated. - ACPX (`acpx_event_scope.rs`, `acpx_provider_state.rs`): an event that names a settled turn of this run becomes a bounded `stale_generation_event` diagnostic (32 per state) and changes no state. Unknown turns still fail. - Codex driver (`codex-session-state.ts`, `codex-session-notifications.ts`): one predicate, `isHistoryTurn()`: the turn is not active, and it was restored from the checkpoint (an earlier generation) or settled in this generation. The classifier's stale-turn branch, the resume-usage branch, restored-terminal replay and the main fail point (`notificationNamesActiveTurn`) all use it. A terminal for an earlier generation's turn is history even when it disagrees with the checkpoint. A terminal that contradicts a turn this same generation settled is still a `conflicting_turn_terminal` integrity failure, and a provider that starts a settled turn again is still rejected. - Transport (`runnerd-codex-transport.ts`): a pending `turn.start` ignores a held failure that names another provider turn. - `DURABLE_TRANSPORT.md` documents the stamp and the rule. ## Verification - `cargo test -p paperclip-runner-core`: new `records_late_events_for_a_settled_turn_as_history` (red before), `stamps_every_event_with_the_runner_process_generation`; two existing ACPX tests now assert "recorded as history, no state change" instead of "rejected". Durable tests: 112 passed. `cargo fmt` clean. - `codex-app-server-driver.stale-generation.test.ts` (new, 9 tests): late item, delta, terminal and workspace change for a settled turn during and before the next turn, an earlier generation's conflicting terminal after recovery, and an unknown turn that still fails. Red on master: 8 of 9 (the during-turn cases on master already dropped the event, with a different diagnostic; the between-turn and recovery cases failed the session). - `runnerd-pending-turn-start-failure.test.ts` (new): an earlier turn's held failure no longer fails the pending start; the pending turn's own failure and a failed run terminal still do. - `vitest run src/drivers/codex src/live/runnerd-codex-transport`: 787 passed; 2 Codex executable-resolution tests timed out under host load and pass when run alone. - `tsc -p tsconfig.json` for the runner package, `check:protocol-types`, `check:replay-goldens`: clean. - Not run: chaos waves (the coordinator schedules them on this host). ## Risks - A late report about a settled turn is now dropped with a diagnostic instead of failing. A provider that keeps working on a settled turn could have its output ignored silently; a provider that starts a settled turn again, or names a turn the runner never saw, still fails closed. - `generation` is a new optional field. Consumers that do not know it ignore it, as the PRP contract requires. - The generation counter is written with the next normal state save, so there is no new write before `turn.start`. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits, test execution). Extended reasoning enabled. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-10 12:37:55 +01:00 · 2026-11-07 23:16:00 -06:00
1 parent f858f79678
commit c0c26188db
@@ -55,17 +55,17 @@
    {
      "path": "schemas/event-v2.schema.json",
      "id": "https://paperclip.dev/schemas/prp/v2/event.schema.json",
-      "sha256": "d7459d9fc186f142b0ea648fd116a2bee52cf7ea71001946be6e2991ccd2ed55"
+      "sha256": "8c1a4264a49b6f8d6713cab666747d2368f57136ebf1c0970a9c67a012bf480a"
    },
    {
      "path": "schemas/event-v3.schema.json",
      "id": "https://paperclip.dev/schemas/prp/v3/event.schema.json",
-      "sha256": "f600e7445e3db04eb394c2d9bb6ab85130f6944b47636372fe9f46f4f8a18939"
+      "sha256": "e357c7b063595e6baa38617541d3b7361c4a54d5d1f8e3711fffc84d660e584e"
    },
    {
      "path": "schemas/event.schema.json",
      "id": "https://paperclip.dev/schemas/prp/v1/event.schema.json",
-      "sha256": "8fd8437758a6e75006a8806bd5e64558393f4d5777febf4e64fbb60c1f4b86e6"
+      "sha256": "ee66887550f3a940c34e91549433da8432ff1050f18e0c0a6c801c41fb82665a"
    },
    {
      "path": "schemas/fixture.schema.json",
@@ -13,6 +13,7 @@
    "sourceSeq": { "type": "integer", "minimum": 1, "maximum": 9007199254740991 },
    "sourceInstanceId": { "type": "string", "minLength": 1, "maxLength": 160 },
    "sourceKind": { "enum": ["runner", "control_plane"] },
+    "generation": { "type": "integer", "minimum": 0, "maximum": 9007199254740991, "description": "The runner process generation that emitted the event. A consumer records an event from an earlier generation for a turn that is not active as history; it never fails the current turn." },
    "runId": { "type": "string", "minLength": 1, "maxLength": 160 },
    "normalizedSessionId": { "type": "string", "minLength": 1, "maxLength": 160 },
    "turnId": { "type": "string", "minLength": 1, "maxLength": 160 },
@@ -41,6 +41,12 @@
        "control_plane"
      ]
    },
+    "generation": {
+      "type": "integer",
+      "minimum": 0,
+      "maximum": 9007199254740991,
+      "description": "The runner process generation that emitted the event. A consumer records an event from an earlier generation for a turn that is not active as history; it never fails the current turn."
+    },
    "runId": {
      "type": "string",
      "minLength": 1,
@@ -22,6 +22,7 @@
    "sourceSeq": { "type": "integer", "minimum": 1, "maximum": 9007199254740991 },
    "sourceInstanceId": { "type": "string", "minLength": 1, "maxLength": 160 },
    "sourceKind": { "enum": ["runner", "control_plane"] },
+    "generation": { "type": "integer", "minimum": 0, "maximum": 9007199254740991, "description": "The runner process generation that emitted the event. A consumer records an event from an earlier generation for a turn that is not active as history; it never fails the current turn." },
    "runId": { "type": "string", "minLength": 1, "maxLength": 160 },
    "normalizedSessionId": { "type": "string", "minLength": 1, "maxLength": 160 },
    "turnId": { "type": "string", "minLength": 1, "maxLength": 240 },
@@ -100,6 +100,25 @@ accepted durable prefix. The ACPX sidecar separately caps queued stdout at
 failure rather than allowing unbounded process memory while the Runner cannot
 read.

+## Event generations and stale events
+
+Every runner event payload carries `generation`: the runner process generation
+that emitted it. A runner process that recovers existing durable state starts
+the next generation. The counter is saved with the next state write, so
+recovery adds no write before the first turn. Events that are still in the
+outbox keep the generation that emitted them. State saved before this field
+existed loads as generation 0.
+
+Consumers apply one rule to provider events. An event that names a turn which
+is not the active turn, and which an earlier generation accepted or this
+generation already settled, is history. The consumer records it and it never
+fails the current turn. This covers late usage, items, workspace changes, and
+terminals for a turn lost on restore. An event for a turn the consumer never
+saw is not history and still fails closed. A terminal that contradicts a turn
+the same generation settled is still a conflict. The ACPX provider state and
+the Codex driver (which Runnerd also uses for OpenCode) apply this rule, and a
+pending `turn.start` ignores a held failure that names another turn.
+
 ## Recovering Codex between turns

 Suppose run A finishes a turn, its Runner exits, and run B wants to continue
@@ -88,12 +88,9 @@ pub enum AcpxEventPayload {
    },
 }

-/// Validates event authority before it decodes the bounded sidecar payload.
-pub fn decode_acpx_event(
-    scope: &AcpxEventScope,
-    event: &AcpxSidecarEvent,
-) -> Result<AcpxEventPayload, LocalRunnerError> {
-    scope.validate_event(event)?;
+/// Applies the payload admission limit. Every sidecar event passes it, the
+/// history of a settled turn included.
+pub fn admit_acpx_event_payload_size(event: &AcpxSidecarEvent) -> Result<(), LocalRunnerError> {
    let payload_bytes = serde_json::to_vec(&event.payload).map_err(|error| {
        LocalRunnerError::invalid(format!("ACPX event payload is invalid: {error}"))
    })?;
@@ -102,6 +99,16 @@ pub fn decode_acpx_event(
            "ACPX event payload exceeds the 256 KiB admission limit",
        ));
    }
+    Ok(())
+}
+
+/// Validates event authority before it decodes the bounded sidecar payload.
+pub fn decode_acpx_event(
+    scope: &AcpxEventScope,
+    event: &AcpxSidecarEvent,
+) -> Result<AcpxEventPayload, LocalRunnerError> {
+    scope.validate_event(event)?;
+    admit_acpx_event_payload_size(event)?;

    match event.event_type {
        GeneratedAcpxSidecarEventType::RuntimeRichEvent => decode_rich_event(&event.payload),
@@ -123,6 +123,19 @@ impl AcpxEventScope {
        Ok(())
    }

+    /// Whether a well-formed event for this run names a turn this scope has
+    /// already settled. Such an event is history from an earlier turn: the
+    /// provider state records it and it never fails the current turn
+    /// (durable-ownership plan, section G). An unknown turn is not history and
+    /// still fails closed in `validate_event`.
+    pub fn names_settled_turn(&self, event: &AcpxSidecarEvent) -> bool {
+        event.run_id.as_deref() == Some(self.run_id.as_str())
+            && event.turn_id.as_deref().is_some_and(|turn_id| {
+                self.active_turn_id.as_deref() != Some(turn_id)
+                    && self.settled_turn_ids.contains(turn_id)
+            })
+    }
+
    pub fn validate_event(&self, event: &AcpxSidecarEvent) -> Result<(), LocalRunnerError> {
        if let Some(run_id) = event.run_id.as_deref() {
            validate_scope_id(run_id, "event run", SHORT_STABLE_ID_CHARS)?;
@@ -3,7 +3,8 @@ use std::collections::{BTreeMap, BTreeSet};
 use serde_json::Value;

 use crate::acpx_event_payload::{
-    decode_acpx_event, AcpxEventPayload, AcpxRuntimeEventKind, AcpxTurnStatus,
+    admit_acpx_event_payload_size, decode_acpx_event, AcpxEventPayload, AcpxRuntimeEventKind,
+    AcpxTurnStatus,
 };
 use crate::acpx_event_scope::AcpxEventScope;
 use crate::acpx_sidecar_transport::AcpxSidecarEvent;
@@ -118,8 +119,13 @@ pub struct AcpxProviderState {
    pending_runtime_request_bytes: usize,
    semantic_result: Option<AcpxSemanticResult>,
    attempted_turn_controls: BTreeSet<String>,
+    history_event_diagnostics: u32,
 }

+/// History diagnostics recorded per provider state before later ones are
+/// dropped silently. History never fails the current turn either way.
+const MAX_HISTORY_EVENT_DIAGNOSTICS: u32 = 32;
+
 impl AcpxProviderState {
    pub fn new(run_id: impl Into<String>) -> Result<Self, LocalRunnerError> {
        Ok(Self {
@@ -137,6 +143,7 @@ impl AcpxProviderState {
            pending_runtime_request_bytes: 0,
            semantic_result: None,
            attempted_turn_controls: BTreeSet::new(),
+            history_event_diagnostics: 0,
        })
    }

@@ -233,6 +240,25 @@ impl AcpxProviderState {
        &mut self,
        event: &AcpxSidecarEvent,
    ) -> Result<Vec<AcpxProviderStateEvent>, LocalRunnerError> {
+        if self.scope.names_settled_turn(event) {
+            // History still passes payload admission like any other event.
+            admit_acpx_event_payload_size(event)?;
+            // A late event for a settled turn (for example a sidecar that
+            // reports the previous turn after the next one started) is
+            // history. Record it; never fail the current turn.
+            self.history_event_diagnostics = self.history_event_diagnostics.saturating_add(1);
+            if self.history_event_diagnostics > MAX_HISTORY_EVENT_DIAGNOSTICS {
+                return Ok(Vec::new());
+            }
+            return Ok(vec![AcpxProviderStateEvent::Diagnostic {
+                code: "stale_generation_event".to_owned(),
+                message: format!(
+                    "recorded a {:?} event for settled turn {} as history",
+                    event.event_type,
+                    event.turn_id.as_deref().unwrap_or_default()
+                ),
+            }]);
+        }
        let payload = decode_acpx_event(&self.scope, event)?;
        match payload {
            AcpxEventPayload::RichActivity {
@@ -1809,6 +1809,43 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                        "tokenUsage": {"total": {"inputTokens": 100, "outputTokens": 20}, "last": {"inputTokens": 50, "outputTokens": 10}}
                    }}))?;
                }
+                // A resumed Codex thread can also report items and the
+                // terminal of the turn lost to a restart after it accepts the
+                // next turn. They are history of an earlier generation.
+                if durable_turn_ids
+                    && state.next_turn > 1
+                    && args
+                        .iter()
+                        .any(|value| value == "--stale-items-after-turn-start")
+                {
+                    let lost_turn_id = format!("provider-turn-{}", state.next_turn - 1);
+                    for (method, params) in [
+                        (
+                            "item/started",
+                            json!({"item": {"id": "lost-item", "type": "agentMessage", "text": ""}}),
+                        ),
+                        (
+                            "item/agentMessage/delta",
+                            json!({"itemId": "lost-item", "delta": "late"}),
+                        ),
+                        (
+                            "item/completed",
+                            json!({"item": {"id": "lost-item", "type": "agentMessage", "text": "late"}}),
+                        ),
+                        (
+                            "turn/completed",
+                            json!({"turn": {"id": lost_turn_id, "status": "failed", "items": [],
+                            "error": {"message": "lost", "codexErrorInfo": "other"}}}),
+                        ),
+                    ] {
+                        let mut params = params;
+                        params["threadId"] = json!(state.thread_id);
+                        if method != "turn/completed" {
+                            params["turnId"] = json!(lost_turn_id);
+                        }
+                        send(json!({"method": method, "params": params}))?;
+                    }
+                }
                // A notification for a turn the runner never owned fails the
                // accepted turn before it starts.
                if args
@@ -525,6 +525,8 @@ pub fn run_durable_runner_with_host_control<E: CommandExecutor>(
        return Ok(());
    }
    if recovered {
+        // A restarted runner process emits a new event generation (section G).
+        state.begin_recovered_generation();
        executor.observe_recovery_state(&state)?;
    }
    if recovered && state.warm_transition.is_none() {
@@ -353,6 +353,13 @@ pub struct DurableState {
    v2_replay_events: BTreeMap<String, StoredV2ReplayEvent>,
    #[serde(default)]
    pub last_connection_protocol_version: Option<u64>,
+    /// The runner process generation that emits events (durable-ownership
+    /// plan, section G). Each runner process that loads existing state starts
+    /// the next generation, and every event envelope carries it, so a consumer
+    /// can record an earlier generation's events as history. State saved
+    /// before this field existed loads as generation 0.
+    #[serde(default)]
+    pub generation: u64,
    pub diagnostics: Vec<String>,
    pub backpressure: bool,
    pub recoverable_failure: Option<String>,
@@ -405,12 +412,20 @@ impl DurableState {
            executor_event_receipts: BTreeMap::new(),
            v2_replay_events: BTreeMap::new(),
            last_connection_protocol_version: None,
+            generation: 1,
            diagnostics: Vec::new(),
            backpressure: false,
            recoverable_failure: None,
        }
    }

+    /// Starts the next event generation for a runner process that recovered
+    /// existing state. It is persisted with the next save, so recovery adds no
+    /// write before the first turn.
+    pub fn begin_recovered_generation(&mut self) {
+        self.generation = self.generation.saturating_add(1);
+    }
+
    pub fn outbox_bytes(&self) -> usize {
        self.outbox.iter().map(|event| event.byte_size).sum()
    }
@@ -638,6 +653,7 @@ impl DurableState {
                "sourceSeq": source_seq,
                "sourceInstanceId": self.runner_instance_id,
                "sourceKind": "runner",
+                "generation": self.generation,
                "runId": self.run_id,
                "normalizedSessionId": self.normalized_session_id,
                "turnId": self.turn_id,
@@ -2859,6 +2875,51 @@ mod tests {
        }
    }

+    // Section G: every runner event carries the runner process generation, so
+    // a consumer can record an earlier generation's events as history.
+    #[test]
+    fn stamps_every_event_with_the_runner_process_generation() {
+        let directory = temporary_directory("event-generation");
+        let _ = fs::remove_dir_all(&directory);
+        let config = config(directory.clone());
+        let store = DurableStateStore::new(&directory).unwrap();
+        let (mut first, recovered) = store.load_or_create(&config).unwrap();
+        assert!(!recovered);
+        assert_eq!(first.generation, 1);
+        first
+            .enqueue_event(&config, "harness.ready", EventPriority::P1, json!({}))
+            .unwrap();
+        assert_eq!(first.outbox[0].envelope["payload"]["generation"], 1);
+        store.save(&first).unwrap();
+
+        // A restarted runner process starts the next generation. Events it
+        // still has to deliver keep the generation that emitted them.
+        let (mut second, recovered) = store.load_or_create(&config).unwrap();
+        assert!(recovered);
+        // Loading alone never changes state; the runner starts the generation.
+        assert_eq!(second.generation, 1);
+        second.begin_recovered_generation();
+        assert_eq!(second.generation, 2);
+        second
+            .enqueue_event(&config, "harness.ready", EventPriority::P1, json!({}))
+            .unwrap();
+        assert_eq!(second.outbox[0].envelope["payload"]["generation"], 1);
+        assert_eq!(second.outbox[1].envelope["payload"]["generation"], 2);
+
+        // State saved before the field existed loads as generation 0, so its
+        // next process is generation 1.
+        let mut legacy = serde_json::to_value(&second).unwrap();
+        legacy.as_object_mut().unwrap().remove("generation");
+        store
+            .save(&serde_json::from_value::<DurableState>(legacy).unwrap())
+            .unwrap();
+        let (mut upgraded, _) = store.load_or_create(&config).unwrap();
+        assert_eq!(upgraded.generation, 0);
+        upgraded.begin_recovered_generation();
+        assert_eq!(upgraded.generation, 1);
+        let _ = fs::remove_dir_all(&directory);
+    }
+
    #[test]
    fn diagnostic_credentials_remain_masked_without_prose_false_positives() {
        let jwt = format!(
@@ -588,14 +588,20 @@ fn terminal_events_retain_dispatched_tools_and_reject_late_turn_events() {
    assert!(state.begin_turn("turn-2").is_err());
    state.complete_tool("call-1", "issues.read").unwrap();
    assert!(state.pending_tool("call-1").is_none());
-    assert!(state
+    // A late event for the settled turn is history: recorded, never applied.
+    let late = state
        .accept_event(&event(
            3,
            GeneratedAcpxSidecarEventType::RuntimeEvent,
            Some("turn-1"),
            json!({"type":"text_delta","text":"late"}),
        ))
-        .is_err());
+        .unwrap();
+    assert!(matches!(
+        late.as_slice(),
+        [AcpxProviderStateEvent::Diagnostic { code, .. }] if code == "stale_generation_event"
+    ));
+    assert_eq!(state.active_turn_id(), None);
 }

 #[test]
@@ -746,14 +752,20 @@ fn terminal_requests_expire_with_their_projected_identity_before_terminal_and_ne
        assert!(state
            .complete_input(projected_id.as_str().unwrap())
            .is_err());
-        assert!(state
+        // A replayed terminal for the settled turn is history and never
+        // replays the terminal or its request expiry.
+        let replay = state
            .accept_event(&event(
                4,
                GeneratedAcpxSidecarEventType::RuntimeTurnTerminal,
                Some("turn-1"),
-                json!({"status":status})
+                json!({"status":status}),
            ))
-            .is_err());
+            .unwrap();
+        assert!(matches!(
+            replay.as_slice(),
+            [AcpxProviderStateEvent::Diagnostic { code, .. }] if code == "stale_generation_event"
+        ));
    }
 }

@@ -816,3 +828,94 @@ fn cursor_plan_parent_identity_survives_decode_pending_and_terminal_projection()
        assert!(state.accept_event(&event(1, GeneratedAcpxSidecarEventType::RuntimeInputRequested, Some("turn-1"), json!({"requestId":"input-1","toolCallId":id,"questionSet":question_set(),"origin":{"provider":"cursor","method":"cursor/create_plan"}}))).is_err());
    }
 }
+
+// Section G: a late sidecar event for a turn that already settled is history.
+// It is recorded as a bounded diagnostic and never fails the current turn,
+// whether it arrives between turns or after the next turn started. An event
+// for an unknown turn still fails closed.
+#[test]
+fn records_late_events_for_a_settled_turn_as_history() {
+    let mut state = AcpxProviderState::new("run-1").unwrap();
+    state.begin_turn("turn-1").unwrap();
+    state
+        .accept_event(&event(
+            1,
+            GeneratedAcpxSidecarEventType::RuntimeTurnTerminal,
+            Some("turn-1"),
+            json!({"status":"completed"}),
+        ))
+        .unwrap();
+    let late = |sequence| {
+        event(
+            sequence,
+            GeneratedAcpxSidecarEventType::RuntimeEvent,
+            Some("turn-1"),
+            json!({"type":"text_delta","text":"late"}),
+        )
+    };
+    let between = state.accept_event(&late(2)).unwrap();
+    assert!(matches!(
+        between.as_slice(),
+        [AcpxProviderStateEvent::Diagnostic { code, .. }] if code == "stale_generation_event"
+    ));
+
+    state.begin_turn("turn-2").unwrap();
+    let during = state.accept_event(&late(3)).unwrap();
+    assert!(matches!(
+        during.as_slice(),
+        [AcpxProviderStateEvent::Diagnostic { code, .. }] if code == "stale_generation_event"
+    ));
+    let late_terminal = state
+        .accept_event(&event(
+            4,
+            GeneratedAcpxSidecarEventType::RuntimeTurnTerminal,
+            Some("turn-1"),
+            json!({"status":"failed"}),
+        ))
+        .unwrap();
+    assert!(matches!(
+        late_terminal.as_slice(),
+        [AcpxProviderStateEvent::Diagnostic { .. }]
+    ));
+    assert_eq!(state.active_turn_id(), Some("turn-2"));
+
+    // The current turn continues normally.
+    let current = state
+        .accept_event(&event(
+            5,
+            GeneratedAcpxSidecarEventType::RuntimeEvent,
+            Some("turn-2"),
+            json!({"type":"text_delta","text":"current"}),
+        ))
+        .unwrap();
+    assert!(!current
+        .iter()
+        .any(|event| matches!(event, AcpxProviderStateEvent::Diagnostic { .. })));
+
+    // History diagnostics are bounded.
+    for sequence in 6..60 {
+        state.accept_event(&late(sequence)).unwrap();
+    }
+    assert!(state.accept_event(&late(59)).unwrap().is_empty());
+
+    // History still passes payload admission.
+    let oversized = state.accept_event(&event(
+        60,
+        GeneratedAcpxSidecarEventType::RuntimeEvent,
+        Some("turn-1"),
+        json!({"type":"text_delta","text":"x".repeat(300 * 1024)}),
+    ));
+    assert!(oversized
+        .unwrap_err()
+        .to_string()
+        .contains("256 KiB admission limit"));
+
+    // A turn this scope never saw is not history.
+    let unknown = state.accept_event(&event(
+        61,
+        GeneratedAcpxSidecarEventType::RuntimeEvent,
+        Some("turn-unknown"),
+        json!({"type":"text_delta","text":"foreign"}),
+    ));
+    assert!(unknown.unwrap_err().to_string().contains("stale turn"));
+}
@@ -4484,13 +4484,36 @@ fn restart_continuation_ignores_late_usage_for_the_turn_lost_on_restore() {
    // lost to the restart after it accepted the continuation turn. The runner
    // must treat that usage as history of a settled turn, not as a foreign
    // event that fails the continuation with turn_binding_mismatch.
-    let directory = temporary_directory("restart-continuation-stale-usage");
+    restart_continuation_ignores_late_events_for_the_lost_turn(
+        "restart-continuation-stale-usage",
+        "--stale-usage-after-turn-start",
+        0,
+    );
+}
+
+#[test]
+fn restart_continuation_ignores_late_items_and_terminal_for_the_turn_lost_on_restore() {
+    // Section G: items and a terminal of the turn an earlier generation lost
+    // are history too. They must not fail the continuation.
+    restart_continuation_ignores_late_events_for_the_lost_turn(
+        "restart-continuation-stale-items",
+        "--stale-items-after-turn-start",
+        4,
+    );
+}
+
+fn restart_continuation_ignores_late_events_for_the_lost_turn(
+    label: &str,
+    stale_flag: &str,
+    expected_history_notices: usize,
+) {
+    let directory = temporary_directory(label);
    let config = provider_config(
        &directory,
        &[
            "--durable-turn-ids",
            "--linger-after-turn-start",
-            "--stale-usage-after-turn-start",
+            stale_flag,
        ],
    );
    let mut first = CodexCommandExecutor::new(&directory);
@@ -4565,6 +4588,19 @@ fn restart_continuation_ignores_late_usage_for_the_turn_lost_on_restore() {
        .any(|event| event.event_type == "run.terminal" || event.event_type == "turn.failed"));
    assert!(events.iter().any(|event| event.event_type == "turn.started"
        && event.payload["providerTurnId"] == "provider-turn-2"));
+    // Positive control: the lost turn's late events reached the runner and
+    // were recorded as history notices.
+    let history_notices = events
+        .iter()
+        .filter(|event| {
+            event.event_type == "provider.notice.recorded"
+                && event.payload["summary"] == "ignored a notification for a settled Codex turn"
+        })
+        .count();
+    assert!(
+        history_notices >= expected_history_notices,
+        "expected {expected_history_notices} history notices, got {history_notices}"
+    );
    assert_eq!(call_count(&directory, "turn/start"), 2);
    restored.shutdown().unwrap();
    fs::remove_dir_all(directory).unwrap();
@@ -0,0 +1,231 @@
+import {
+  FakeCodexTransport,
+  WORKSPACE,
+  describe,
+  expect,
+  it,
+  makeDriver,
+  vi,
+  type PrpEvent,
+} from "./codex-app-server-driver.test-support.js";
+
+// Section G of the durable-ownership plan: an event that names a turn which
+// is not the active turn, and which an earlier session generation accepted or
+// this generation already settled, is history. It is recorded and never fails
+// the current turn. Runnerd drives Codex, OpenCode and ACPX providers through
+// this driver, so the rule covers all three. A turn the session never saw
+// still fails closed (codex-app-server-driver.terminal-replay.test.ts).
+
+async function settledThenActive(runId: string, options: { betweenTurns?: boolean } = {}) {
+  const transport = new FakeCodexTransport();
+  const session = await makeDriver([transport], { conversationMode: "direct" }).openSession({
+    runId,
+    normalizedSessionId: `normalized-${runId}`,
+    workingDirectory: WORKSPACE,
+  });
+  const events: PrpEvent[] = [];
+  const consume = (async () => {
+    for await (const event of session.events()) events.push(event);
+  })();
+  await session.startTurn({ message: { role: "user", text: "First." } });
+  transport.push("turn/started", { threadId: "thread-1", turn: { id: "turn-1", status: "inProgress", items: [] } });
+  transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] } });
+  await vi.waitFor(() => expect(events.some((event) =>
+    event.eventType === "turn.completed" && event.turnId === "turn-1")).toBe(true));
+  // A late event can arrive before the next turn starts.
+  const pushHistory = (method: string, params: Record<string, unknown>) => transport.push(method, { threadId: "thread-1", ...params });
+  if (options.betweenTurns) return { transport, session, events, consume, pushHistory };
+  transport.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } });
+  await session.startTurn({ message: { role: "user", text: "Second." } });
+  transport.push("turn/started", { threadId: "thread-1", turn: { id: "turn-2", status: "inProgress", items: [] } });
+  await vi.waitFor(() => expect(events.some((event) =>
+    event.eventType === "turn.started" && event.turnId === "turn-2")).toBe(true));
+  return { transport, session, events, consume, pushHistory };
+}
+
+describe("Codex driver stale-generation events", () => {
+  it.each([
+    ["an item start", "item/started", { turnId: "turn-1", item: { id: "late-1", type: "agentMessage", text: "" } }],
+    ["an item completion", "item/completed", { turnId: "turn-1", item: { id: "late-1", type: "agentMessage", text: "late" } }],
+    ["an item update", "item/agentMessage/delta", { turnId: "turn-1", itemId: "late-1", delta: "late" }],
+    ["a repeated terminal", "turn/completed", { turn: { id: "turn-1", status: "failed", items: [],
+      error: { code: "provider_failed", message: "late", recoverable: false } } }],
+    ["a workspace change", "paperclip/workspaceChange/updated", { turnId: "turn-1", changes: [] }],
+  ] as const)("records %s for a settled turn as history while the next turn runs", async (_label, method, params) => {
+    const { transport, session, events, consume } = await settledThenActive(`stale-${method.replace(/\W+/g, "-")}`);
+    transport.push(method, { threadId: "thread-1", ...params });
+    transport.push("item/completed", { threadId: "thread-1", turnId: "turn-2",
+      item: { id: "current-1", type: "agentMessage", text: "Current work." } });
+    transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } });
+    await vi.waitFor(() => expect(events.some((event) =>
+      event.eventType === "turn.completed" && event.turnId === "turn-2")).toBe(true));
+    await session.close({ reason: "test complete" });
+    await consume;
+    expect(events.find((event) => event.eventType === "session.failed")).toBeUndefined();
+    expect(events.filter((event) => event.eventType === "turn.failed")).toEqual([]);
+    expect(events.filter((event) => event.turnId === "turn-1" && event.eventType !== "harness.diagnostic")
+      .map((event) => event.eventType)).toEqual(["turn.accepted", "turn.started", "turn.completed"]);
+    expect(events.find((event) => event.eventType === "harness.diagnostic"
+      && event.payload.classification === "stale_generation_event")?.payload)
+      .toMatchObject({ receivedTurnId: "turn-1", expectedTurnId: "turn-2" });
+  });
+
+  it.each([
+    ["an item completion", "item/completed", { turnId: "turn-1", item: { id: "late-1", type: "agentMessage", text: "late" } }],
+    ["an item update", "item/agentMessage/delta", { turnId: "turn-1", itemId: "late-1", delta: "late" }],
+    ["a tool start", "item/started", { turnId: "turn-1",
+      item: { id: "cmd-1", type: "commandExecution", command: "ls", status: "inProgress" } }],
+    ["a tool completion", "item/completed", { turnId: "turn-1",
+      item: { id: "cmd-1", type: "commandExecution", command: "ls", status: "completed" } }],
+    ["a plan update", "turn/plan/updated", { turnId: "turn-1", plan: [{ step: "Inspect", status: "completed" }] }],
+  ] as const)("records %s for a settled turn as history before the next turn starts", async (_label, method, params) => {
+    const { transport, session, events, consume, pushHistory } = await settledThenActive(
+      `between-${method.replace(/\W+/g, "-")}`, { betweenTurns: true });
+    pushHistory(method, params);
+    await vi.waitFor(() => expect(events.some((event) => event.eventType === "harness.diagnostic"
+      && event.payload.classification === "stale_generation_event")).toBe(true));
+    transport.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } });
+    await session.startTurn({ message: { role: "user", text: "Second." } });
+    transport.push("turn/started", { threadId: "thread-1", turn: { id: "turn-2", status: "inProgress", items: [] } });
+    transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } });
+    await vi.waitFor(() => expect(events.some((event) =>
+      event.eventType === "turn.completed" && event.turnId === "turn-2")).toBe(true));
+    await session.close({ reason: "test complete" });
+    await consume;
+    expect(events.find((event) => event.eventType === "session.failed")).toBeUndefined();
+    expect(events.find((event) => event.eventType === "harness.diagnostic"
+      && event.payload.classification === "stale_generation_event")?.payload)
+      .toMatchObject({ receivedTurnId: "turn-1", expectedTurnId: null });
+    // Only a history diagnostic: the old activity never re-enters the stream.
+    expect(events.filter((event) => event.eventType.startsWith("tool.execution.")
+      || event.eventType === "plan.updated" || (event.turnId === "turn-1" && event.eventType.startsWith("item.")))).toEqual([]);
+  });
+
+  // A recovered session's checkpointed turns belong to an earlier
+  // generation. Their late reports, including a terminal that disagrees with
+  // the checkpoint, are history. (A terminal that contradicts a turn this
+  // generation settled is still a conflict: semantic-results tests.)
+  it("records an earlier generation's late reports as history after recovery", async () => {
+    const first = new FakeCodexTransport();
+    const second = new FakeCodexTransport();
+    second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE,
+      turns: [{ id: "turn-1", status: "completed", items: [] }] } };
+    const driver = makeDriver([first, second], { conversationMode: "direct" });
+    const original = await driver.openSession({ runId: "run-prior-generation",
+      normalizedSessionId: "normalized-prior-generation", workingDirectory: WORKSPACE });
+    const originalEvents = (async () => { for await (const _event of original.events()) { /* drain */ } })();
+    await original.startTurn({ message: { role: "user", text: "First." } });
+    first.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] } });
+    await vi.waitFor(async () => expect((await original.snapshot()).terminalTurns).toEqual([
+      expect.objectContaining({ turnId: "turn-1" })]));
+    const snapshot = await original.snapshot();
+    await original.close({ reason: "controller restarted" });
+    await originalEvents;
+
+    const recovery = await driver.recoverSession?.(snapshot);
+    const recovered = recovery?.session;
+    if (!recovered) throw new Error("recovery did not return a session");
+    await recovered.reconcile?.();
+    const events: PrpEvent[] = [];
+    const consume = (async () => { for await (const event of recovered.events()) events.push(event); })();
+    second.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "failed", items: [],
+      error: { code: "provider_turn_lost_on_restore", message: "lost", recoverable: false } } });
+    second.push("item/completed", { threadId: "thread-1", turnId: "turn-1",
+      item: { id: "late-1", type: "agentMessage", text: "late" } });
+    second.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } });
+    await recovered.startTurn({ message: { role: "user", text: "Second." } });
+    second.push("turn/started", { threadId: "thread-1", turn: { id: "turn-2", status: "inProgress", items: [] } });
+    second.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } });
+    await vi.waitFor(() => expect(events.some((event) =>
+      event.eventType === "turn.completed" && event.turnId === "turn-2")).toBe(true));
+    await recovered.close({ reason: "test complete" });
+    await consume;
+    expect(events.find((event) => event.eventType === "session.failed")).toBeUndefined();
+    expect(events.filter((event) => event.eventType === "turn.failed")).toEqual([]);
+    expect(events.some((event) => event.eventType === "harness.diagnostic"
+      && ["stale_generation_event", "restored_terminal_replay"].includes(String(event.payload.classification))
+      && event.payload.receivedTurnId === "turn-1")).toBe(true);
+  });
+
+  // A restore-loss report for a turn this session never bound records that
+  // turn as an earlier generation's. Its later activity is history too.
+  it("records late activity for a turn reported lost on restore as history", async () => {
+    const transport = new FakeCodexTransport();
+    const session = await makeDriver([transport]).openSession({
+      runId: "run-restore-lost-activity",
+      normalizedSessionId: "normalized-restore-lost-activity",
+      workingDirectory: WORKSPACE,
+    });
+    await session.startTurn({ message: { role: "user", text: "Continue." } });
+    transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-uncheckpointed", status: "failed",
+      items: [], error: { code: "provider_turn_lost_on_restore", message: "lost", recoverable: true } } });
+    transport.push("item/completed", { threadId: "thread-1", turnId: "turn-uncheckpointed",
+      item: { id: "late-1", type: "agentMessage", text: "late" } });
+    transport.push("item/agentMessage/delta", { threadId: "thread-1", turnId: "turn-uncheckpointed",
+      itemId: "late-1", delta: "late" });
+    transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] } });
+    const events: PrpEvent[] = [];
+    for await (const event of session.events()) {
+      events.push(event);
+      if (event.eventType === "turn.completed" || event.eventType === "session.failed") break;
+    }
+    expect(events.find((event) => event.eventType === "session.failed")).toBeUndefined();
+    expect(events.find((event) => event.eventType === "turn.completed")?.turnId).toBe("turn-1");
+    expect(events.filter((event) => event.eventType === "harness.diagnostic"
+      && event.payload.classification === "stale_generation_event"
+      && event.payload.receivedTurnId === "turn-uncheckpointed").length).toBeGreaterThan(0);
+  });
+
+  // The checkpoint's active turn belongs to the earlier generation too. When
+  // reconcile settles it from thread/read, a later, different terminal for it
+  // is history, not a conflicting terminal of this generation.
+  it("records a late terminal for a checkpointed active turn that reconcile settled as history", async () => {
+    const first = new FakeCodexTransport();
+    const second = new FakeCodexTransport();
+    second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE,
+      turns: [{ id: "turn-1", status: "completed", items: [] }] } };
+    const driver = makeDriver([first, second], { conversationMode: "direct" });
+    const original = await driver.openSession({ runId: "run-active-checkpoint",
+      normalizedSessionId: "normalized-active-checkpoint", workingDirectory: WORKSPACE });
+    const originalEvents = (async () => { for await (const _event of original.events()) { /* drain */ } })();
+    await original.startTurn({ message: { role: "user", text: "First." } });
+    first.push("turn/started", { threadId: "thread-1", turn: { id: "turn-1", status: "inProgress", items: [] } });
+    await vi.waitFor(async () => expect((await original.snapshot()).activeTurnId).toBe("turn-1"));
+    const snapshot = await original.snapshot();
+    await original.close({ reason: "controller restarted" });
+    await originalEvents;
+
+    const recovery = await driver.recoverSession?.(snapshot);
+    const recovered = recovery?.session;
+    if (!recovered) throw new Error("recovery did not return a session");
+    const events: PrpEvent[] = [];
+    const consume = (async () => { for await (const event of recovered.events()) events.push(event); })();
+    await recovered.reconcile?.();
+    await vi.waitFor(() => expect(events.some((event) =>
+      event.turnId === "turn-1" && event.eventType === "turn.completed")).toBe(true));
+    second.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "failed", items: [],
+      error: { code: "provider_failed", message: "late", recoverable: false } } });
+    second.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } });
+    await recovered.startTurn({ message: { role: "user", text: "Second." } });
+    second.push("turn/started", { threadId: "thread-1", turn: { id: "turn-2", status: "inProgress", items: [] } });
+    second.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } });
+    await vi.waitFor(() => expect(events.some((event) =>
+      event.eventType === "turn.completed" && event.turnId === "turn-2")).toBe(true));
+    await recovered.close({ reason: "test complete" });
+    await consume;
+    expect(events.find((event) => event.eventType === "session.failed")).toBeUndefined();
+    expect(events.some((event) => event.eventType === "harness.diagnostic"
+      && event.payload.classification === "stale_generation_event"
+      && event.payload.receivedTurnId === "turn-1")).toBe(true);
+  });
+
+  it("still fails closed on a turn the session never saw", async () => {
+    const { transport, session, events, consume } = await settledThenActive("stale-unknown");
+    transport.push("item/completed", { threadId: "thread-1", turnId: "turn-foreign",
+      item: { id: "foreign-1", type: "agentMessage", text: "foreign" } });
+    await consume;
+    await session.close({ reason: "test complete" }).catch(() => undefined);
+    expect(events.find((event) => event.eventType === "session.failed")?.payload)
+      .toMatchObject({ code: "turn_binding_mismatch" });
+  });
+});
@@ -22,6 +22,7 @@ import {
  admitResult,
  captureResultFromItem,
  mapTerminalTurn,
+  settledTerminalReplayConflict,
 } from "./codex-session-terminal.js";
 import {
  recordCanonicalWorkspaceChange,
@@ -53,6 +54,11 @@ const DELTA_KINDS: Record<string, string> = {

 // Turn-scoped activity a resumed runner can replay from committed history.
 // turn/completed is excluded: its replay still passes the fingerprint check.
+/** Methods that keep their own handling for a turn the history rule covers. */
+const HISTORY_RULE_EXEMPT_METHODS: ReadonlySet<string> = new Set([
+  "turn/completed", "turn/started", "thread/tokenUsage/updated", "paperclip/canonicalProviderEvent",
+]);
+
 const REPLAYABLE_TURN_METHODS = new Set([
  "turn/started",
  "item/started",
@@ -174,14 +180,14 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex
    const identity = classifyCodexNotification({
      method: notification.method, params: notification.params, runId: state.runId,
      rootThreadId: state.opened.threadId, activeTurnId: state.activeTurnId,
-      knownThreads: new Set(state.lineageByThread.keys()), settledTurns: new Set(state.terminalTurns.keys()),
+      knownThreads: new Set(state.lineageByThread.keys()), settledTurns: state.historyTurnIds(),
    });
    if ((identity.classification === "root" || identity.classification === "stale_turn")
      && identity.threadId === state.opened.threadId
      && notification.method === "thread/tokenUsage/updated"
      && identity.turnId !== null
      && identity.turnId !== state.activeTurnId
-      && (state.activeTurnId === null || (identity.turnId !== null && state.terminalTurns.has(identity.turnId)))) {
+      && (state.activeTurnId === null || state.isHistoryTurn(identity.turnId))) {
      state.codexUsageBaseline = observeCodexUsage(state.codexUsageBaseline, record(notification.params.tokenUsage).total, true);
      state.usageSnapshot = codexRunUsage(state.codexUsageBaseline);
      if (state.notificationIdentityDiagnostics++ < 32) state.emit("harness.diagnostic", {
@@ -190,6 +196,22 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex
      });
      return;
    }
+    if (identity.classification === "stale_turn") {
+      // History by the one stale-event rule (CodexSessionState.isHistoryTurn).
+      state.recordHistoryEvent(notification.method, identity.turnId ?? "");
+      return;
+    }
+    // Apply the rule before any canonical mapping, so a settled turn's late
+    // items and plans never re-enter the stream as fresh work. Its terminal
+    // keeps its own conflict check below, a provider that starts it again
+    // still fails, and canonical provider diagnostics keep flowing.
+    if (identity.classification === "root" && identity.turnId !== null
+      && identity.threadId === state.opened.threadId
+      && state.isHistoryTurn(identity.turnId)
+      && !HISTORY_RULE_EXEMPT_METHODS.has(notification.method)) {
+      state.recordHistoryEvent(notification.method, identity.turnId);
+      return;
+    }
    if (identity.classification !== "root") {
      if (state.notificationIdentityDiagnostics < 32) {
        state.notificationIdentityDiagnostics += 1;
@@ -543,6 +565,16 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex
      // channel. Wait for the pending turn/start to settle first, so
      // turn.accepted always precedes the terminal event for the same turn.
      await state.turnStartSettled;
+      // A terminal for a turn an earlier generation accepted is history, even
+      // when it disagrees with the checkpointed terminal (for example, a
+      // restored provider that reports the turn lost on restore). A terminal
+      // that contradicts a turn this generation settled is still a conflict.
+      if (state.priorGenerationTurns.has(turnId) && state.isHistoryTurn(turnId)) {
+        const settled = state.terminalTurns.get(turnId);
+        if (settled === undefined || settledTerminalReplayConflict(state, turn, turn, settled) !== null)
+          state.recordHistoryEvent(notification.method, turnId);
+        return;
+      }
      if (state.terminalTurns.has(turnId)) {
        mapTerminalTurn(state, turn, turnId);
        return;
@@ -169,6 +169,12 @@ export class CodexSessionState {
   * notifications; those replays are not new provider activity.
   */
  readonly restoredTerminalTurns = new Set<string>();
+  /**
+   * Turns that an earlier session generation accepted: every turn this
+   * session restored from its checkpoint. With the turns settled in this
+   * generation they form the history set (see isHistoryTurn).
+   */
+  readonly priorGenerationTurns = new Set<string>();
  readonly workspaceChangesByTurn = new Map<string, Record<string, unknown>>();
  readonly emittedFileReferences = new Set<string>();
  #workspaceRoots: string[] | null = null;
@@ -331,6 +337,15 @@ export class CodexSessionState {
    this.dispositionOnlyRecoveryConsumed =
      dispositionOnlyRecoveryPreviouslyConsumed;
    this.dispositionOnlyRecoveryTurnId = dispositionOnlyRecoveryTurnId;
+    for (const turnId of this.terminalTurns.keys()) this.priorGenerationTurns.add(turnId);
+    // The checkpoint's active turn (or one the previous controller's runner
+    // accepted) also belongs to an earlier generation. isHistoryTurn keeps it
+    // live while it stays active; once reconcile settles it, its late reports
+    // are history too.
+    for (const turnId of [input.activeTurnId, input.adoptedAcceptedTurnId])
+      if (turnId) this.priorGenerationTurns.add(turnId);
+    if (dispositionOnlyRecoveryTurnId && dispositionOnlyRecoveryTurnId !== this.activeTurnId)
+      this.priorGenerationTurns.add(dispositionOnlyRecoveryTurnId);
  }

  requireActiveTurn(turnId: string, operation: string): void {
@@ -379,13 +394,47 @@ export class CodexSessionState {
   * fingerprint was already committed, and turn/completed still checks it.
   */
  isRestoredTerminalReplay(turnId: string): boolean {
+    return this.restoredTerminalTurns.has(turnId) && this.isHistoryTurn(turnId);
+  }
+
+  /**
+   * The one stale-event rule (durable-ownership plan, section G). A provider
+   * event that names a turn which is not the active turn, and which an earlier
+   * session generation accepted (restored from the checkpoint, or reported
+   * lost on restore) or this generation already settled, is history: it is
+   * recorded and never fails the current turn. A turn this session never saw
+   * is not history and still fails closed.
+   */
+  isHistoryTurn(turnId: string): boolean {
    return turnId.length > 0
      && !this.protocolFailed
      && turnId !== this.activeTurnId
-      && this.restoredTerminalTurns.has(turnId);
+      && (this.priorGenerationTurns.has(turnId) || this.restoredTerminalTurns.has(turnId)
+        || this.terminalTurns.has(turnId));
+  }
+
+  /** Every turn the history rule covers, for classifiers that take a set. */
+  historyTurnIds(): Set<string> {
+    return new Set([...this.priorGenerationTurns, ...this.restoredTerminalTurns, ...this.terminalTurns.keys()]);
+  }
+
+  recordHistoryEvent(kind: string, turnId: string): void {
+    if (this.notificationIdentityDiagnostics >= 32) return;
+    this.notificationIdentityDiagnostics += 1;
+    this.emit("harness.diagnostic", {
+      code: "provider_notification_identity",
+      classification: "stale_generation_event",
+      kind: kind.slice(0, 128),
+      expectedTurnId: this.activeTurnId,
+      receivedTurnId: turnId.slice(0, 256),
+    });
  }

  notificationNamesActiveTurn(turnId: string, kind: string): boolean {
+    if (this.isHistoryTurn(turnId)) {
+      this.recordHistoryEvent(kind, turnId);
+      return false;
+    }
    if (
      this.protocolFailed ||
      this.terminal ||
@@ -893,16 +893,24 @@ const TURN_START_ADOPTION_CHECK_RESERVE_MS = 5_000;
 * The runner's own final failure of a turn whose start is still pending: a
 * non-recoverable turn terminal held for that start, or a failed run terminal.
 * Neither can be followed by a live start, so the start reports it at once.
+ * A turn failure that names another provider turn (for example, the turn an
+ * earlier generation lost on restore) is history, never this start's failure
+ * (durable-ownership plan, section G).
 */
 export function runnerdPendingTurnStartFailure(
  events: ReadonlyArray<{ eventType: string; envelope: { payload?: unknown } }>,
+  expectedProviderTurnId: string | null = null,
 ): string | null {
  for (const event of events) {
    const payload = record(record(event.envelope.payload).payload);
    const error = record(payload.error);
    if (
      event.eventType === "turn.failed" &&
-      (payload.recoverable === false || error.recoverable === false)
+      (payload.recoverable === false || error.recoverable === false) &&
+      terminalMayBelongToPendingTurnStart({
+        expectedProviderTurnId,
+        terminalProviderTurnId: runnerdTerminalProviderTurnId(event.eventType, payload),
+      })
    ) {
      const code = typeof error.code === "string" ? error.code
        : typeof payload.code === "string" ? payload.code : "provider_turn_failed";
@@ -7280,7 +7288,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
          ...(this.#core?.store.state.committedEvents ?? []).filter(
            (event) => event.sourceSeq > this.#eventSourceSeq,
          ),
-        ]);
+        ], expectedProviderTurnId);
        if (failure !== null)
          throw new Error(`runnerd failed the provider turn before it started: ${failure}`);
      };
@@ -0,0 +1,27 @@
+import { describe, expect, it } from "vitest";
+import { runnerdPendingTurnStartFailure } from "./runnerd-codex-transport.js";
+
+// Section G: a held turn failure that names another provider turn (for
+// example, the turn an earlier generation lost on restore) is history. It must
+// never fail the turn.start that is pending now (#19722 reported any held
+// failure).
+const failed = (providerTurnId: string | undefined, code = "provider_failed") => ({
+  eventType: "turn.failed",
+  envelope: { payload: { payload: { ...(providerTurnId ? { providerTurnId } : {}),
+    recoverable: false, error: { code, message: "failed", recoverable: false } } } },
+});
+
+describe("runnerdPendingTurnStartFailure", () => {
+  it("ignores a held failure of an earlier turn", () => {
+    expect(runnerdPendingTurnStartFailure([failed("turn-lost", "provider_turn_lost_on_restore")], "turn-new")).toBeNull();
+  });
+  it("reports the pending turn's own failure", () => {
+    expect(runnerdPendingTurnStartFailure([failed("turn-lost"), failed("turn-new")], "turn-new"))
+      .toBe("provider_failed: failed");
+  });
+  it("reports a failure that names no turn, and a failed run terminal", () => {
+    expect(runnerdPendingTurnStartFailure([failed(undefined)], "turn-new")).toBe("provider_failed: failed");
+    expect(runnerdPendingTurnStartFailure([{ eventType: "run.terminal",
+      envelope: { payload: { payload: { runTerminalState: "failed" } } } }], "turn-new")).toBe("run_terminal_failed");
+  });
+});
@@ -4971,6 +4971,12 @@ export const eventSchema = {
        "control_plane"
      ]
    },
+    "generation": {
+      "type": "integer",
+      "minimum": 0,
+      "maximum": 9007199254740991,
+      "description": "The runner process generation that emitted the event. A consumer records an event from an earlier generation for a turn that is not active as history; it never fails the current turn."
+    },
    "runId": {
      "type": "string",
      "minLength": 1,
@@ -5754,6 +5760,12 @@ export const eventV2Schema = {
        "control_plane"
      ]
    },
+    "generation": {
+      "type": "integer",
+      "minimum": 0,
+      "maximum": 9007199254740991,
+      "description": "The runner process generation that emitted the event. A consumer records an event from an earlier generation for a turn that is not active as history; it never fails the current turn."
+    },
    "runId": {
      "type": "string",
      "minLength": 1,
@@ -6089,6 +6101,12 @@ export const eventV3Schema = {
        "control_plane"
      ]
    },
+    "generation": {
+      "type": "integer",
+      "minimum": 0,
+      "maximum": 9007199254740991,
+      "description": "The runner process generation that emitted the event. A consumer records an event from an earlier generation for a turn that is not active as history; it never fails the current turn."
+    },
    "runId": {
      "type": "string",
      "minLength": 1,
@@ -4,4 +4,4 @@
 import equalModule from "ajv/dist/runtime/equal.js";
 import ucs2LengthModule from "ajv/dist/runtime/ucs2length.js";

-"use strict";export const authCredentialsValidator: unknown = validate20;const schema31 = {"$schema":"https://json-schema.org/draft/2020-12/schema","$id":"https://paperclip.dev/schemas/prp/v1/auth-credentials.schema.json","title":"PRP credential update notification","description":"Content-free evidence of changed credentials for the harness-auth binding in this normalized session. Credential-store detection does not prove an OAuth refresh succeeded.","type":"object","required":["schema","provider","bindingId","revision","detection"],"properties":{"schema":{"const":"paperclip.auth.credentials.updated.v1"},"provider":{"type":"string","minLength":1,"maxLength":160,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:-]*$"},"bindingId":{"type":"string","minLength":1,"maxLength":160,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:-]*$"},"revision":{"type":"integer","minimum":1,"maximum":9007199254740991},"detection":{"enum":["credential_store","native_hook"]}},"additionalProperties":false};const func1 = typeof ucs2LengthModule === "function" ? ucs2LengthModule : ucs2LengthModule.default;const pattern4 = new RegExp("^[A-Za-z0-9][A-Za-z0-9._:-]*$", "u");function validate20(data, {instancePath="", parentData, parentDataProperty, rootData=data, dynamicAnchors={}}={}){/*# sourceURL="https://paperclip.dev/schemas/prp/v1/auth-credentials.schema.json" */;let vErrors = null;let errors = 0;const evaluated0 = validate20.evaluated;if(evaluated0.dynamicProps){evaluated0.props = undefined;}if(evaluated0.dynamicItems){evaluated0.items = undefined;}if(data && typeof data == "object" && !Array.isArray(data)){if(data.schema === undefined){const err0 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "schema"},message:"must have required property '"+"schema"+"'"};if(vErrors === null){vErrors = [err0];}else {vErrors.push(err0);}errors++;}if(data.provider === undefined){const err1 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "provider"},message:"must have required property '"+"provider"+"'"};if(vErrors === null){vErrors = [err1];}else {vErrors.push(err1);}errors++;}if(data.bindingId === undefined){const err2 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "bindingId"},message:"must have required property '"+"bindingId"+"'"};if(vErrors === null){vErrors = [err2];}else {vErrors.push(err2);}errors++;}if(data.revision === undefined){const err3 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "revision"},message:"must have required property '"+"revision"+"'"};if(vErrors === null){vErrors = [err3];}else {vErrors.push(err3);}errors++;}if(data.detection === undefined){const err4 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "detection"},message:"must have required property '"+"detection"+"'"};if(vErrors === null){vErrors = [err4];}else {vErrors.push(err4);}errors++;}for(const key0 in data){if(!(((((key0 === "schema") || (key0 === "provider")) || (key0 === "bindingId")) || (key0 === "revision")) || (key0 === "detection"))){const err5 = {instancePath,schemaPath:"#/additionalProperties",keyword:"additionalProperties",params:{additionalProperty: key0},message:"must NOT have additional properties"};if(vErrors === null){vErrors = [err5];}else {vErrors.push(err5);}errors++;}}if(data.schema !== undefined){if("paperclip.auth.credentials.updated.v1" !== data.schema){const err6 = {instancePath:instancePath+"/schema",schemaPath:"#/properties/schema/const",keyword:"const",params:{allowedValue: "paperclip.auth.credentials.updated.v1"},message:"must be equal to constant"};if(vErrors === null){vErrors = [err6];}else {vErrors.push(err6);}errors++;}}if(data.provider !== undefined){let data1 = data.provider;if(typeof data1 === "string"){if(func1(data1) > 160){const err7 = {instancePath:instancePath+"/provider",schemaPath:"#/properties/provider/maxLength",keyword:"maxLength",params:{limit: 160},message:"must NOT have more than 160 characters"};if(vErrors === null){vErrors = [err7];}else {vErrors.push(err7);}errors++;}if(func1(data1) < 1){const err8 = {instancePath:instancePath+"/provider",schemaPath:"#/properties/provider/minLength",keyword:"minLength",params:{limit: 1},message:"must NOT have fewer than 1 characters"};if(vErrors === null){vErrors = [err8];}else {vErrors.push(err8);}errors++;}if(!pattern4.test(data1)){const err9 = {instancePath:instancePath+"/provider",schemaPath:"#/properties/provider/pattern",keyword:"pattern",params:{pattern: "^[A-Za-z0-9][A-Za-z0-9._:-]*$"},message:"must match pattern \""+"^[A-Za-z0-9][A-Za-z0-9._:-]*$"+"\""};if(vErrors === null){vErrors = [err9];}else {vErrors.push(err9);}errors++;}}else {const err10 = {instancePath:instancePath+"/provider",schemaPath:"#/properties/provider/type",keyword:"type",params:{type: "string"},message:"must be string"};if(vErrors === null){vErrors = [err10];}else {vErrors.push(err10);}errors++;}}if(data.bindingId !== undefined){let data2 = data.binLine truncated
+"use strict";export const authCredentialsValidator: unknown = validate20;const schema31 = {"$schema":"https://json-schema.org/draft/2020-12/schema","$id":"https://paperclip.dev/schemas/prp/v1/auth-credentials.schema.json","title":"PRP credential update notification","description":"Content-free evidence of changed credentials for the harness-auth binding in this normalized session. Credential-store detection does not prove an OAuth refresh succeeded.","type":"object","required":["schema","provider","bindingId","revision","detection"],"properties":{"schema":{"const":"paperclip.auth.credentials.updated.v1"},"provider":{"type":"string","minLength":1,"maxLength":160,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:-]*$"},"bindingId":{"type":"string","minLength":1,"maxLength":160,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:-]*$"},"revision":{"type":"integer","minimum":1,"maximum":9007199254740991},"detection":{"enum":["credential_store","native_hook"]}},"additionalProperties":false};const func1 = typeof ucs2LengthModule === "function" ? ucs2LengthModule : ucs2LengthModule.default;const pattern4 = new RegExp("^[A-Za-z0-9][A-Za-z0-9._:-]*$", "u");function validate20(data, {instancePath="", parentData, parentDataProperty, rootData=data, dynamicAnchors={}}={}){/*# sourceURL="https://paperclip.dev/schemas/prp/v1/auth-credentials.schema.json" */;let vErrors = null;let errors = 0;const evaluated0 = validate20.evaluated;if(evaluated0.dynamicProps){evaluated0.props = undefined;}if(evaluated0.dynamicItems){evaluated0.items = undefined;}if(data && typeof data == "object" && !Array.isArray(data)){if(data.schema === undefined){const err0 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "schema"},message:"must have required property '"+"schema"+"'"};if(vErrors === null){vErrors = [err0];}else {vErrors.push(err0);}errors++;}if(data.provider === undefined){const err1 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "provider"},message:"must have required property '"+"provider"+"'"};if(vErrors === null){vErrors = [err1];}else {vErrors.push(err1);}errors++;}if(data.bindingId === undefined){const err2 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "bindingId"},message:"must have required property '"+"bindingId"+"'"};if(vErrors === null){vErrors = [err2];}else {vErrors.push(err2);}errors++;}if(data.revision === undefined){const err3 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "revision"},message:"must have required property '"+"revision"+"'"};if(vErrors === null){vErrors = [err3];}else {vErrors.push(err3);}errors++;}if(data.detection === undefined){const err4 = {instancePath,schemaPath:"#/required",keyword:"required",params:{missingProperty: "detection"},message:"must have required property '"+"detection"+"'"};if(vErrors === null){vErrors = [err4];}else {vErrors.push(err4);}errors++;}for(const key0 in data){if(!(((((key0 === "schema") || (key0 === "provider")) || (key0 === "bindingId")) || (key0 === "revision")) || (key0 === "detection"))){const err5 = {instancePath,schemaPath:"#/additionalProperties",keyword:"additionalProperties",params:{additionalProperty: key0},message:"must NOT have additional properties"};if(vErrors === null){vErrors = [err5];}else {vErrors.push(err5);}errors++;}}if(data.schema !== undefined){if("paperclip.auth.credentials.updated.v1" !== data.schema){const err6 = {instancePath:instancePath+"/schema",schemaPath:"#/properties/schema/const",keyword:"const",params:{allowedValue: "paperclip.auth.credentials.updated.v1"},message:"must be equal to constant"};if(vErrors === null){vErrors = [err6];}else {vErrors.push(err6);}errors++;}}if(data.provider !== undefined){let data1 = data.provider;if(typeof data1 === "string"){if(func1(data1) > 160){const err7 = {instancePath:instancePath+"/provider",schemaPath:"#/properties/provider/maxLength",keyword:"maxLength",params:{limit: 160},message:"must NOT have more than 160 characters"};if(vErrors === null){vErrors = [err7];}else {vErrors.push(err7);}errors++;}if(func1(data1) < 1){const err8 = {instancePath:instancePath+"/provider",schemaPath:"#/properties/provider/minLength",keyword:"minLength",params:{limit: 1},message:"must NOT have fewer than 1 characters"};if(vErrors === null){vErrors = [err8];}else {vErrors.push(err8);}errors++;}if(!pattern4.test(data1)){const err9 = {instancePath:instancePath+"/provider",schemaPath:"#/properties/provider/pattern",keyword:"pattern",params:{pattern: "^[A-Za-z0-9][A-Za-z0-9._:-]*$"},message:"must match pattern \""+"^[A-Za-z0-9][A-Za-z0-9._:-]*$"+"\""};if(vErrors === null){vErrors = [err9];}else {vErrors.push(err9);}errors++;}}else {const err10 = {instancePath:instancePath+"/provider",schemaPath:"#/properties/provider/type",keyword:"type",params:{type: "string"},message:"must be string"};if(vErrors === null){vErrors = [err10];}else {vErrors.push(err10);}errors++;}}if(data.bindingId !== undefined){let data2 = data.binLine truncated