diff --git a/apps/presentation/dashboard/src/features/personal-workspace/research-usage-selection.test.mjs b/apps/presentation/dashboard/src/features/personal-workspace/research-usage-selection.test.mjs
index 50b8188cec..69089d1015 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/research-usage-selection.test.mjs
+++ b/apps/presentation/dashboard/src/features/personal-workspace/research-usage-selection.test.mjs
@@ -37,6 +37,19 @@ test('unknown, partial and measured zero remain distinct', () => {
   assert.equal(selectResearchUsage(zero, 13, 13).turns_without_counters, 0);
 });
 
+test('range selection includes an interrupted Turn before later completed work', () => {
+  const capture = {...usage, turns: [turn(1), turn(3)], incomplete_turn: null,
+    interrupted_turns: [{turn_id: 'interrupted', start: turn(2).start, observed_through: turn(2).end,
+      first_line: 6, last_line: 8, counter_observations: 1, usage: turn(2, 40).usage}]};
+  const result = selectResearchUsage(capture, 1, 3);
+  assert.deepEqual(result.turns.map(row => row.turn_id), ['turn-1', 'interrupted', 'turn-3']);
+  assert.equal(result.turns[1].incomplete, true);
+  assert.equal(result.completed_turns, 2);
+  assert.equal(result.observed_input_tokens, 240);
+  assert.equal(result.observed_turn_seconds, 30);
+  assert.equal(selectResearchUsage(capture, 2, 2).observed_input_tokens, 40);
+});
+
 test('rejects empty, reversed, fractional and out-of-capture selections', () => {
   for (const [first, last] of [[0,1],[2,1],[1,14],[1.5,3],[1,NaN],[1,Infinity]]) {
     assert.throws(() => selectResearchUsage(usage, first, last), /valid captured Turn range/);
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/research-usage-selection.ts b/apps/presentation/dashboard/src/features/personal-workspace/research-usage-selection.ts
index 7e55de42c2..f500e4223f 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/research-usage-selection.ts
+++ b/apps/presentation/dashboard/src/features/personal-workspace/research-usage-selection.ts
@@ -2,14 +2,14 @@ import type {TrajectorySolverUsage} from "../../../../../../loopx/control_plane/
 
 /** Presentation arithmetic over one already-qualified capture, not a ledger. */
 export function researchUsageRows(usage: TrajectorySolverUsage) {
-  const rows: Array<Omit<TrajectorySolverUsage["turns"][number], "usage"> & {
-    usage: TrajectorySolverUsage["turns"][number]["usage"] | null; ordinal: number; incomplete: boolean;
-  }> = usage.turns.map((turn, index) => ({...turn, ordinal: index + 1, incomplete: false}));
-  const tail = usage.incomplete_turn;
-  if (tail) rows.push({...tail, end: tail.observed_through,
-    elapsed_seconds: Math.max(0, (Date.parse(tail.observed_through) - Date.parse(tail.start)) / 1000),
-    ordinal: rows.length + 1, incomplete: true, usage: tail.usage});
-  return rows;
+  const complete = usage.turns.map(turn => ({...turn, incomplete: false}));
+  const partial = [...(usage.interrupted_turns ?? [])];
+  if (usage.incomplete_turn) partial.push(usage.incomplete_turn);
+  return [...complete, ...partial.map(turn => ({...turn, end: turn.observed_through,
+    elapsed_seconds: Math.max(0, (Date.parse(turn.observed_through) - Date.parse(turn.start)) / 1000),
+    incomplete: true}))]
+    .sort((left, right) => left.first_line - right.first_line)
+    .map((turn, index) => ({...turn, ordinal: index + 1}));
 }
 
 export function selectResearchUsage(usage: TrajectorySolverUsage, first: number, last: number) {
diff --git a/benchmark/runtime/provider_request_admission.py b/benchmark/runtime/provider_request_admission.py
index eff68ac58b..de03bb9961 100644
--- a/benchmark/runtime/provider_request_admission.py
+++ b/benchmark/runtime/provider_request_admission.py
@@ -88,7 +88,9 @@ def __init__(self, env, session_wake, receipt, wake, *, timeout=None, original_q
             "session_id": self._invocation.session_id, "requests": 0,
             "pending": 0, "sent": 0, "unknown": 0, "refused": 0,
         }
-        receipt.setdefault("provider_request_invocations", []).append(self._summary)
+        self._receipt_invocations = receipt.setdefault("provider_request_invocations", [])
+        self._receipt_index = len(self._receipt_invocations)
+        self._receipt_invocations.append(self._summary)
 
     def started(self, process):
         with self._lock:
@@ -99,7 +101,12 @@ def started(self, process):
 
     def retire(self):
         with self._lock:
+            if self._retired:
+                return
             self._retired = True
+            # Freeze the invocation receipt at its first exit. A gateway
+            # handler can still append observations while cleanup drains.
+            self._receipt_invocations[self._receipt_index] = self._summary.copy()
             self._ready.set()
 
     def before_cleanup(self, _process, _scope_error):
diff --git a/benchmark/tests/test_provider_request_admission.py b/benchmark/tests/test_provider_request_admission.py
index 7a2b3bf064..5325dd23bd 100644
--- a/benchmark/tests/test_provider_request_admission.py
+++ b/benchmark/tests/test_provider_request_admission.py
@@ -405,14 +405,19 @@ def observed_connect(connection):
                     old_event, = _finished(original_wake, invocation_id=old_summary["invocation_id"])
                     new_event, = _finished(successor_wake, invocation_id=new_summary["invocation_id"])
                     assert old_event["turn_id"] == "old" and old_event["forwarded"] is False
+                    assert old_event["refusal_code"] == "provider_request_without_live_owner"
                     assert new_event["turn_id"] == ("old" if shared_receipt else "new")
                     assert new_event["forwarded"] is True
                     assert old_event["session_id"] == "session-old" and new_event["session_id"] == "session-new"
                     assert old_event["invocation_id"] != new_event["invocation_id"]
-                    assert old_summary["sent"] == 0 and old_summary["refused"] == 1
+                    # Only the upstream-connect case had parsed and observed
+                    # the full request before retirement. The late refusal
+                    # remains in the journal without rewriting the snapshot.
+                    observed_at_exit = int(stage == "upstream_connect")
+                    assert old_summary["sent"] == old_summary["refused"] == 0
+                    assert old_summary["requests"] == old_summary["pending"] == observed_at_exit
                     assert new_summary["sent"] == 1 and new_summary["refused"] == 0
-                    assert old_summary["requests"] == new_summary["requests"] == 1
-                    assert old_summary["pending"] == new_summary["pending"] == 0
+                    assert new_summary["requests"] == 1 and new_summary["pending"] == 0
                     assert "refusal_code" not in new_summary
                     if shared_receipt:
                         assert original_receipt is successor_receipt and original_wake == successor_wake
diff --git a/docs/reference/trajectory-assessment.md b/docs/reference/trajectory-assessment.md
index fdc375952b..67fbcfd8f4 100644
--- a/docs/reference/trajectory-assessment.md
+++ b/docs/reference/trajectory-assessment.md
@@ -1042,33 +1042,36 @@ checkpoint reads are qualified separately from this presentation example.
 ![Scoped interpretation and usage reader](../development/assets/research-analysis/summary.png)
 
 
-### Observe an unfinished final Turn
+### Observe usage across an interruption
 
-For a captured run that stopped mid-Turn, the curator may explicitly add
+For a capture with an unfinished or interrupted Turn, the curator may add
 `--include-incomplete-usage` to the same registration command, together with
 `--usage-jsonl` and `--usage-session-id`. This reads persisted counters without
 running a model or changing the solver. The default complete-session contract
 above remains unchanged.
 
 The opt-in method is `codex_observed_session_counters_v1`. Its totals include
-observed counters in the final unfinished Turn. `completed_turns` and `turns`
-still contain only measured completions; `incomplete_turn` identifies the final
-open interval, source lines, observation count and observed delta. With no
-counter in that interval its `usage` is null, never an invented zero. With no
-measured counters anywhere registration fails. Duplicate snapshots do not add
-usage. Both methods reject wrong identity, forks, missing prefixes, gaps, resets,
-unmatched completions and malformed JSONL. Cancellation without a completion
-remains incomplete; recovery that starts another Turn without closing the prior
-one remains unsupported. This opt-in is not a relaxed integrity qualification.
-
-The Research workbench's existing Sources and usage disclosure shows the
-unfinished interval and its persisted input/output next to the capture totals,
-with complete cost unknown. No extra owner confirmation or model run is needed
-to read these facts. An unfinished interval ends at the last supported observed
-event, not at an inferred process death. Observed time includes tools and waits;
-unpersisted requests, later activity, model prices, evaluator costs and operator
-attention remain unknown. Do not sum overlapping captures or repeated pack
-registrations: the pack is a versioned diagnostic, not a billing ledger.
+persisted counters in interrupted Turns and the final open interval. A native
+`turn_aborted` event with the active Turn's ID closes that interval for
+observation. A later Turn can then contribute its own counters in the same
+session. `interrupted_turns` preserves the interruption boundary, source lines
+and observed delta. `incomplete_turn` describes an interval still open at the
+end of the capture. Both record `usage: null` when the interval has no counters.
+`completed_turns` and `turns` contain measured completions only.
+
+Registration fails if the capture has no measured counters. Both methods reject
+wrong identities, forks, missing prefixes, counter gaps or resets, and malformed
+JSONL. Starting a new Turn still requires a matching prior terminal event.
+An abort without a Turn ID cannot close that interval. The complete-session mode
+rejects a captured interruption.
+
+In Research → Sources and usage → **Find time and input hotspots**, interrupted
+intervals appear in chronological order as incomplete. Their recorded counts
+participate in the selected range, and **Counter evidence** shows the verified
+native boundary. Observed time includes tools and waits. Unpersisted requests,
+later activity, model prices, evaluator costs and operator attention remain
+unknown. Overlapping captures and repeated registrations can include the same
+usage. Each pack is a versioned diagnostic.
 
 The original source hash and historical/unavailable handling also apply to this
 mode. Select a capture-free or default complete-session pack to stop using it;
diff --git a/examples/personal-workspace-browser/research-turn-costs.mjs b/examples/personal-workspace-browser/research-turn-costs.mjs
index 95c7a07f8c..23582e2d0a 100644
--- a/examples/personal-workspace-browser/research-turn-costs.mjs
+++ b/examples/personal-workspace-browser/research-turn-costs.mjs
@@ -36,7 +36,7 @@ for i in range(10):
  totals={key:totals[key]+counts[key] for key in totals}
  seconds+=120 if i==6 else 1
  append('event_msg',dict(type='token_count',info=dict(total_token_usage=totals.copy(),last_token_usage=counts)))
- seconds+=1;append('event_msg',dict(type='task_complete',turn_id=f'turn-{i+1}'))
+ seconds+=1;append('event_msg',dict(type='turn_aborted' if i==6 else 'task_complete',turn_id=f'turn-{i+1}'))
 seconds+=1;append('event_msg',dict(type='task_started',turn_id='unfinished'))
 analysis=dict(title='Observed session',input=pack,artifacts={'facts.txt':text,'usage.jsonl':'\\n'.join(map(json.dumps,rows))},outputs={},usage_jsonl='usage.jsonl',usage_session_id='session',include_incomplete_usage=True)
 Path(sys.argv[1]).write_text(json.dumps(dict(goal_id=goal,display_name='Synthetic cost study',events=[build_explore_node_event(goal_id=goal,node_id='question',title=text,recorded_at='2026-01-01T00:00:00Z')],analyses=[analysis])))
@@ -60,11 +60,12 @@ Path(sys.argv[1]).write_text(json.dumps(dict(goal_id=goal,display_name='Syntheti
       assert.match(await items.first().innerText(), /3,000/);
       assert.match(await items.first().innerText(), /2,000/);
       await costs.getByRole('combobox').selectOption('elapsed');
-      assert.match(await items.first().innerText(), /第 7 轮/);
+      assert.match(await items.first().innerText(), /第 7 轮.*未完成/s);
       await items.first().getByRole('button').click();
       const source = page.getByRole('region', {name: '引用原文', exact: true});
       await source.locator('pre').waitFor();
       assert.match(await source.innerText(), /turn-7/);
+      assert.match(await source.innerText(), /turn_aborted/);
       assert.match(await source.innerText(), /token_count/);
       assert.ok(!(await source.innerText()).includes('SENSITIVE_TEXT_NOT_FOR_BROWSER'));
       await source.focus(); await page.keyboard.press('Escape');
@@ -82,7 +83,7 @@ Path(sys.argv[1]).write_text(json.dumps(dict(goal_id=goal,display_name='Syntheti
       const firstTurn = costs.getByRole('spinbutton', {name: '起始轮次', exact: true});
       const lastTurn = costs.getByRole('spinbutton', {name: '结束轮次', exact: true});
       const selection = costs.locator('.research-usage-range-summary');
-      assert.match(await selection.innerText(), /已选 11 \/ 11 轮.*完成 10 轮.*无计数 1 轮/);
+      assert.match(await selection.innerText(), /已选 11 \/ 11 轮.*完成 9 轮.*无计数 1 轮/);
       const download = async () => {
         const waiting = page.waitForEvent('download');
         await costs.getByRole('button', {name: '下载所选用量', exact: true}).click();
@@ -92,6 +93,7 @@ Path(sys.argv[1]).write_text(json.dumps(dict(goal_id=goal,display_name='Syntheti
       const all = await download();
       assert.equal(all.selection.turns.length, 11, 'Download includes every selected Turn beyond the eight-row page');
       assert.equal(all.selection.observed_input_tokens, 5900);
+      assert.equal(all.selection.turns.find(turn => turn.turn_id === 'turn-7').incomplete, true);
       assert.equal(all.goal_id, 'synthetic-cost-study');
       assert.equal(all.artifact_sha256.length, 64);
       assert.ok(!JSON.stringify(all).includes('SENSITIVE_TEXT_NOT_FOR_BROWSER'));
@@ -140,6 +142,10 @@ Path(sys.argv[1]).write_text(json.dumps(dict(goal_id=goal,display_name='Syntheti
         assert.equal(await helpButton.evaluate(element => element === document.activeElement), true);
         await costs.locator('legend').evaluate(element => element.scrollIntoView({block: 'center'}));
         await page.screenshot({path: resolve(outputDir, `research-usage-range-${width}.png`), animations: 'disabled'});
+        await costs.getByRole('combobox').selectOption('elapsed');
+        await items.first().evaluate(element => element.scrollIntoView({block: 'center'}));
+        await page.screenshot({path: resolve(outputDir, `research-interrupted-turn-${width}.png`), animations: 'disabled'});
+        await costs.getByRole('combobox').selectOption('input');
       }
       await page.evaluate(() => localStorage.setItem('loopx-pw-locale', 'en'));
       await page.reload({waitUntil: 'networkidle'});
@@ -150,7 +156,7 @@ Path(sys.argv[1]).write_text(json.dumps(dict(goal_id=goal,display_name='Syntheti
       await costs.locator('summary').click();
       await costs.getByRole('spinbutton', {name: 'First Turn', exact: true}).fill('3');
       await costs.getByRole('spinbutton', {name: 'Last Turn', exact: true}).fill('7');
-      assert.match(await selection.innerText(), /5 \/ 11 Turns selected.*5 completed.*0 without counters/);
+      assert.match(await selection.innerText(), /5 \/ 11 Turns selected.*4 completed.*0 without counters/);
       assert.match(await selection.innerText(), /5,400/);
       const englishDownload = page.waitForEvent('download');
       await costs.getByRole('button', {name: 'Download selected usage', exact: true}).click();
diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
index 2cdd368158..45e7219808 100644
--- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
+++ b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
@@ -181,6 +181,7 @@ def _usage_event(line_number: int, event: dict) -> dict | None:
     elif event.get("type") == "event_msg" and payload.get("type") in {
         "task_started",
         "task_complete",
+        "turn_aborted",
         "token_count",
     }:
         payload = {
@@ -787,13 +788,14 @@ def read_usage_turn(*, runtime_root: Path, owner_goal_id: str, pack_id: str,
     if not captured:
         raise PackUnavailable("usage_not_captured")
     projection = captured["projection"]
-    turns = [*projection["turns"]]
+    turns = [(turn, "task_complete") for turn in projection["turns"]]
+    turns.extend((turn, "turn_aborted") for turn in projection.get("interrupted_turns", []))
     if projection.get("incomplete_turn"):
-        turns.append(projection["incomplete_turn"])
-    matches = [turn for turn in turns if turn["turn_id"] == turn_id]
+        turns.append((projection["incomplete_turn"], None))
+    matches = [(turn, end_kind) for turn, end_kind in turns if turn["turn_id"] == turn_id]
     if len(matches) != 1:
         raise PackUnavailable("usage_turn_not_captured")
-    turn = matches[0]
+    turn, end_kind = matches[0]
     first, last = turn["first_line"], turn["last_line"]
     digest = hashlib.sha256()
     baseline = start = counter = end = None
@@ -815,7 +817,7 @@ def read_usage_turn(*, runtime_root: Path, owner_goal_id: str, pack_id: str,
                 start = observed
             elif kind == "token_count" and observed["payload"].get("info"):
                 counter = observed
-            elif kind == "task_complete":
+            elif line == last and kind == end_kind and observed["payload"].get("turn_id") == turn_id:
                 end = observed
         if digest.hexdigest() != captured["artifact_sha256"]:
             raise PackUnavailable("source_changed")
diff --git a/loopx/cli_commands/benchmark_trajectory.py b/loopx/cli_commands/benchmark_trajectory.py
index a1d0580968..15532c63d8 100644
--- a/loopx/cli_commands/benchmark_trajectory.py
+++ b/loopx/cli_commands/benchmark_trajectory.py
@@ -56,7 +56,7 @@ def register_benchmark_trajectory_commands(subparsers, add_format) -> None:
     )
     register.add_argument(
         "--include-incomplete-usage", action="store_true",
-        help="Observe persisted counters through an unfinished final Turn; requires usage capture. Does not establish complete cost or solver termination.",
+        help="Observe persisted counters from interrupted or unfinished Turns. Requires a usage capture. Unpersisted usage remains unknown.",
     )
     register.add_argument(
         "--native-outcomes-json",
diff --git a/loopx/control_plane/capabilities/trajectory_assessment_types.ts b/loopx/control_plane/capabilities/trajectory_assessment_types.ts
index 706a3fa0df..a13fc5de1c 100644
--- a/loopx/control_plane/capabilities/trajectory_assessment_types.ts
+++ b/loopx/control_plane/capabilities/trajectory_assessment_types.ts
@@ -82,11 +82,17 @@ export interface TrajectoryContinuation {
 }
 
 /** Provider-observed facts, separate from inferred episode judgments. */
+export interface TrajectoryPartialTurnUsage {
+  turn_id: string; start: string; observed_through: string; first_line: number; last_line: number;
+  counter_observations: number; usage: {input_tokens: number; cached_input_tokens: number;
+    uncached_input_tokens: number; output_tokens: number} | null;
+}
+
 export interface TrajectorySolverUsage {
   session_id: string; method: 'codex_complete_session_counters_v1' | 'codex_observed_session_counters_v1';
-  incomplete_turn?: {turn_id: string; start: string; observed_through: string; first_line: number; last_line: number;
-    counter_observations: number; usage: {input_tokens: number; cached_input_tokens: number;
-      uncached_input_tokens: number; output_tokens: number} | null} | null;
+  incomplete_turn?: TrajectoryPartialTurnUsage | null;
+  /** Exact native abort boundaries. Usage remains limited to persisted counters. */
+  interrupted_turns?: TrajectoryPartialTurnUsage[];
   start: string; end: string; completed_turns: number; counter_observations: number;
   input_tokens: number; cached_input_tokens: number; uncached_input_tokens: number; output_tokens: number;
   elapsed_seconds: number; turn_elapsed_seconds: number;
diff --git a/loopx/control_plane/capabilities/trajectory_codex_usage.ts b/loopx/control_plane/capabilities/trajectory_codex_usage.ts
index 4d0cfbbe22..17c1a96b16 100644
--- a/loopx/control_plane/capabilities/trajectory_codex_usage.ts
+++ b/loopx/control_plane/capabilities/trajectory_codex_usage.ts
@@ -1,7 +1,7 @@
 /** Read-only Codex session adapter for benchmark-toolkit, separate from model judgments. */
 import {EffectRuntimeRequestError} from '../effect_runtime_errors.ts';
-import type {TrajectorySolverUsage} from './trajectory_assessment_types.ts';
-import {reconcileTrajectoryRequests, summarizeTrajectoryLoopXInput} from './trajectory_request_usage.ts';
+import type {TrajectoryPartialTurnUsage, TrajectorySolverUsage} from './trajectory_assessment_types.ts';
+import {isIdentifiedTurnAbort, reconcileTrajectoryRequests, summarizeTrajectoryLoopXInput} from './trajectory_request_usage.ts';
 import {estimateTrajectoryLoopXInput} from './trajectory_input_usage.ts';
 import {summarizeTrajectoryToolContext} from './trajectory_tool_context.ts';
 import {summarizeTrajectoryLoopXRequests} from './trajectory_loopx_requests.ts';
@@ -21,15 +21,27 @@ const time = (v: unknown): number => {
   if (typeof v !== 'string' || !/^\d{4}-\d\d-\d\dT.*(?:Z|[+-]\d\d:\d\d)$/.test(v) || !Number.isFinite(Date.parse(v))) return fail('timestamp must include timezone');
   return Date.parse(v);
 };
+type ActiveTurn = {turn_id: string; start: string; start_ms: number; first_line: number; baseline: Counters; samples: number};
+
+function partialTurn(turn: ActiveTurn, observedThrough: string, lastLine: number, total: Counters): TrajectoryPartialTurnUsage {
+  const usage = subtract(total, turn.baseline);
+  if (usage.cached_input_tokens > usage.input_tokens) return fail('invalid delta cache accounting');
+  return {turn_id: turn.turn_id, start: turn.start, observed_through: observedThrough,
+    first_line: turn.first_line, last_line: lastLine, counter_observations: turn.samples,
+    usage: turn.samples ? {...usage, uncached_input_tokens: usage.input_tokens - usage.cached_input_tokens} : null};
+}
+
 export function summarizeTrajectoryCodexUsage(params: Record<string, unknown>): TrajectorySolverUsage {
   if (!Array.isArray(params.events) || typeof params.session_id !== 'string' || !params.session_id) return fail('session events and expected identity required');
   if (params.include_incomplete !== undefined && typeof params.include_incomplete !== 'boolean') return fail('include_incomplete must be boolean');
   const observed = params.include_incomplete === true;
   const rows = params.events;
-  let observedThrough = '', observedLine = 0;
+  let start = '', terminalThrough = '', observedThrough = '', observedLine = 0;
   let sessionSeen = false, previousTime = -Infinity, previous = zero(), samples = 0;
-  let active: {turn_id: string; start: string; start_ms: number; first_line: number; baseline: Counters; samples: number} | null = null;
-  const turns: Array<{turn_id: string; start: string; end: string; first_line: number; last_line: number; elapsed_seconds: number; usage: Counters & {uncached_input_tokens: number}}> = [];
+  let active: ActiveTurn | null = null, interruptedSeconds = 0;
+  const turns: TrajectorySolverUsage['turns'] = [];
+  const interruptedTurns: TrajectoryPartialTurnUsage[] = [];
+  const started = new Set<string>();
   let previousLine = 0;
   for (const event of rows) {
     if (!event || typeof event !== 'object' || Array.isArray(event)) return fail('invalid event');
@@ -42,14 +54,25 @@ export function summarizeTrajectoryCodexUsage(params: Record<string, unknown>):
       continue;
     }
     if (!sessionSeen) return fail('session header missing');
-    if (event.type !== 'event_msg' || !['task_started', 'task_complete', 'token_count'].includes(p?.type)) continue;
+    if (event.type !== 'event_msg') continue;
+    const interrupted = isIdentifiedTurnAbort(p);
+    if (!interrupted && !['task_started', 'task_complete', 'token_count'].includes(p?.type)) continue;
     const stamp = time(event.timestamp);
     if (stamp < previousTime) return fail('out-of-order events');
     previousTime = stamp;
     observedThrough = event.timestamp; observedLine = event.line;
     if (p.type === 'task_started') {
-      if (active || typeof p.turn_id !== 'string' || !p.turn_id || turns.some(t => t.turn_id === p.turn_id)) return fail('duplicate or overlapping Turn');
+      if (active || typeof p.turn_id !== 'string' || !p.turn_id || started.has(p.turn_id)) return fail('duplicate or overlapping Turn');
+      start ||= event.timestamp;
+      started.add(p.turn_id);
       active = {turn_id: p.turn_id, start: event.timestamp, start_ms: stamp, first_line: event.line, baseline: {...previous}, samples: 0};
+    } else if (interrupted) {
+      if (!active || p.turn_id !== active.turn_id) return fail('unmatched aborted Turn');
+      if (!observed) return fail('capture includes an aborted Turn; use observed usage');
+      interruptedTurns.push(partialTurn(active, observedThrough, observedLine, previous));
+      interruptedSeconds += (stamp - active.start_ms) / 1000;
+      terminalThrough = event.timestamp;
+      active = null;
     } else if (p.type === 'token_count' && p.info) {
       if (!active) return fail('usage outside a captured Turn');
       const total = counters(p.info.total_token_usage), last = counters(p.info.last_token_usage);
@@ -67,17 +90,14 @@ export function summarizeTrajectoryCodexUsage(params: Record<string, unknown>):
       turns.push({turn_id: active.turn_id, start: active.start, end: event.timestamp, first_line: active.first_line,
         last_line: event.line, elapsed_seconds: (stamp - active.start_ms) / 1000,
         usage: {...usage, uncached_input_tokens: usage.input_tokens - usage.cached_input_tokens}});
+      terminalThrough = event.timestamp;
       active = null;
     }
   }
   if (!observed && (active || !turns.length)) return fail('capture must end after a completed Turn');
   if (!samples) return fail('no measured usage in captured session');
-  const tail = active ? subtract(previous, active.baseline) : null;
-  if (tail && tail.cached_input_tokens > tail.input_tokens) return fail('invalid delta cache accounting');
-  const incomplete = active ? {turn_id: active.turn_id, start: active.start, observed_through: observedThrough,
-    first_line: active.first_line, last_line: observedLine, counter_observations: active.samples,
-    usage: active.samples && tail ? {...tail, uncached_input_tokens: tail.input_tokens - tail.cached_input_tokens} : null} : null;
-  const start = turns[0]?.start ?? active!.start, end = active ? observedThrough : turns.at(-1)!.end;
+  const incomplete = active ? partialTurn(active, observedThrough, observedLine, previous) : null;
+  const end = active ? observedThrough : terminalThrough;
   const requestCoverage = reconcileTrajectoryRequests(rows, params.session_id);
   const loopxInput = summarizeTrajectoryLoopXInput(rows, params.session_id, requestCoverage)
     ?? estimateTrajectoryLoopXInput(params.loopx_input_trace, params.session_id, params.input_goal_id, rows, requestCoverage);
@@ -91,11 +111,13 @@ export function summarizeTrajectoryCodexUsage(params: Record<string, unknown>):
     ...(toolContext ? {tool_context: toolContext} : {}),
     ...(loopxRequests ? {loopx_requests: loopxRequests} : {}),
     ...(observed ? {incomplete_turn: incomplete} : {}),
+    ...(interruptedTurns.length ? {interrupted_turns: interruptedTurns} : {}),
     start, end, completed_turns: turns.length, counter_observations: samples,
     input_tokens: previous.input_tokens, cached_input_tokens: previous.cached_input_tokens,
     uncached_input_tokens: previous.input_tokens - previous.cached_input_tokens, output_tokens: previous.output_tokens,
     elapsed_seconds: (time(end) - time(start)) / 1000,
-    turn_elapsed_seconds: turns.reduce((sum, t) => sum + t.elapsed_seconds, 0) + (active ? (time(end) - active.start_ms) / 1000 : 0), turns,
+    turn_elapsed_seconds: turns.reduce((sum, t) => sum + t.elapsed_seconds, interruptedSeconds)
+      + (active ? (time(end) - active.start_ms) / 1000 : 0), turns,
     limitations: [observed
       ? 'Observed counters only, including any unfinished final Turn; unpersisted requests and later usage remain unknown. No termination or complete cost is inferred.'
       : 'Complete captured session only; donor/prefix study costs and evaluator costs are not included.',
diff --git a/loopx/control_plane/capabilities/trajectory_loopx_requests.ts b/loopx/control_plane/capabilities/trajectory_loopx_requests.ts
index 4722eabede..33dd75f64b 100644
--- a/loopx/control_plane/capabilities/trajectory_loopx_requests.ts
+++ b/loopx/control_plane/capabilities/trajectory_loopx_requests.ts
@@ -3,6 +3,7 @@
  * Turn. This conservative order association is not provider input attribution,
  * pure-control classification, complete workflow cost or a savings estimate. */
 import type {TrajectoryLoopXRequests, TrajectoryRequestCoverage, TrajectoryRequestTokens} from './trajectory_assessment_types.ts';
+import {isIdentifiedTurnAbort} from './trajectory_request_usage.ts';
 import {trajectoryLoopXOperation} from './trajectory_tool_context.ts';
 
 type Row = {line: number; type?: string; payload?: Record<string, any>};
@@ -24,7 +25,8 @@ export function summarizeTrajectoryLoopXRequests(rows: Row[], tools: unknown,
   try {
     if (!Array.isArray(tools) || coverage?.status !== 'verified') throw new Error('Reconciled request and tool observations required.');
     const ordered = [...rows.filter(r => r.type === 'token_usage_record'
-      || r.type === 'event_msg' && ['task_started', 'task_complete'].includes(r.payload?.type)),
+      || r.type === 'event_msg' && (['task_started', 'task_complete'].includes(r.payload?.type)
+        || isIdentifiedTurnAbort(r.payload))),
       ...tools.map(r => ({...r, tool: true}))].sort((a, b) => a.line - b.line);
     const calls = new Map<string, Call>(), seen = new Set<string>(), outputs = new Set<string>();
     const compactions = new Set(rows.filter(r => r.type === 'compacted').map(r => r.payload?.compaction_response_id));
diff --git a/loopx/control_plane/capabilities/trajectory_request_usage.ts b/loopx/control_plane/capabilities/trajectory_request_usage.ts
index 60b730ff8d..5ff9def702 100644
--- a/loopx/control_plane/capabilities/trajectory_request_usage.ts
+++ b/loopx/control_plane/capabilities/trajectory_request_usage.ts
@@ -29,6 +29,10 @@ function record(value: unknown, session: string) {
     usage: tokens(p.usage), thread: tokens(p.thread_token_usage), turn: tokens(p.turn_token_usage)};
 }
 type Request = ReturnType<typeof record>;
+
+export function isIdentifiedTurnAbort(payload: {type?: unknown; turn_id?: unknown} | undefined | null): boolean {
+  return payload?.type === 'turn_aborted' && typeof payload.turn_id === 'string' && payload.turn_id.length > 0;
+}
 const same = (a: Request, b: Request) => a.response_id === b.response_id && a.turn_id === b.turn_id
   && a.root_turn_id === b.root_turn_id && equal(a.usage, b.usage) && equal(a.thread, b.thread) && equal(a.turn, b.turn);
 
@@ -54,10 +58,12 @@ export function reconcileTrajectoryRequests(rows: Array<{type?: unknown; payload
     for (const e of rows) {
       const p = object(e.payload);
       if (e.type === 'session_meta') continue;
-      if (e.type !== 'token_usage_record' && e.type !== 'compacted'
+      const interrupted = e.type === 'event_msg' && isIdentifiedTurnAbort(p);
+      if (e.type !== 'token_usage_record' && e.type !== 'compacted' && !interrupted
         && !(e.type === 'event_msg' && ['task_started', 'task_complete', 'token_count'].includes(p.type as string))) continue;
       if (e.type === 'event_msg' && p.type === 'task_started') active = p.turn_id as string;
-      if (e.type === 'event_msg' && p.type === 'task_complete') active = null;
+      if (interrupted && p.turn_id !== active) return invalid('Aborted Turn identity mismatch.');
+      if (interrupted || e.type === 'event_msg' && p.type === 'task_complete') active = null;
       if (typeof e.timestamp !== 'string' || !/\dT.*(?:Z|[+-]\d\d:\d\d)$/.test(e.timestamp)
         || !Number.isFinite(Date.parse(e.timestamp)) || Date.parse(e.timestamp) < previousTime) return invalid('Request timestamps are invalid or out of order.');
       previousTime = Date.parse(e.timestamp);
diff --git a/tests/capabilities/test_provider_gateway_observation.py b/tests/capabilities/test_provider_gateway_observation.py
index bd751f3b22..f2e21156b0 100644
--- a/tests/capabilities/test_provider_gateway_observation.py
+++ b/tests/capabilities/test_provider_gateway_observation.py
@@ -9,6 +9,8 @@
 from types import SimpleNamespace
 from urllib.parse import urlsplit
 
+import pytest
+
 from benchmark.runtime.provider_request_admission import ProviderRequestAdmission
 
 
@@ -153,6 +155,48 @@ def record(event):
         assert "synthetic-provider-key" not in json.dumps(events)
 
 
+@pytest.mark.parametrize("response_during_cleanup", [False, True])
+def test_invocation_exit_keeps_pending_snapshot_after_late_response(
+    tmp_path, monkeypatch, response_during_cleanup,
+):
+    with _waiting_upstream() as (upstream, received, release, _observed):
+        guard, receipt = _guard(tmp_path, upstream, _Admission())
+        dispatched = threading.Event()
+        finished = threading.Event()
+        observe = guard.observe
+
+        def record(event):
+            observe(event)
+            if event["event"] == "dispatch":
+                dispatched.set()
+            elif event["event"] == "finished":
+                finished.set()
+
+        monkeypatch.setattr(guard, "observe", record)
+        shutdown = ThreadingHTTPServer.shutdown
+
+        def finish_during_gateway_shutdown(server):
+            if server.server_port == urlsplit(guard.base_url).port:
+                release.set()
+                assert finished.wait(timeout=5)
+            shutdown(server)
+
+        with guard.serve(["codex", "exec", "-"]):
+            client, _result = _request(guard.base_url)
+            assert received.wait(timeout=5) and dispatched.wait(timeout=5)
+            guard.before_cleanup(None, None)
+            if response_during_cleanup:
+                monkeypatch.setattr(ThreadingHTTPServer, "shutdown", finish_during_gateway_shutdown)
+        [snapshot] = receipt["provider_request_invocations"]
+        assert snapshot["requests"] == snapshot["sent"] == snapshot["pending"] == 1
+        try:
+            release.set()
+            assert finished.wait(timeout=5)
+            assert receipt["provider_request_invocations"][0]["pending"] == 1
+        finally:
+            client.join(timeout=5)
+
+
 def test_blocked_request_log_does_not_hold_invocation_or_admission_lock(tmp_path, monkeypatch):
     log_blocked = threading.Event()
     release_log = threading.Event()
diff --git a/tests/capabilities/test_trajectory_assessment_store.py b/tests/capabilities/test_trajectory_assessment_store.py
index b5b7d64caf..5a90bd5bc0 100644
--- a/tests/capabilities/test_trajectory_assessment_store.py
+++ b/tests/capabilities/test_trajectory_assessment_store.py
@@ -1323,9 +1323,22 @@ def test_observed_usage_preserves_incomplete_cost_and_default_gate(pack):
     assert inspect_pack(**scope)["solver_usage"]["reason_code"] == "source_unavailable"
 
 
-def test_observed_usage_real_cli_without_model_dispatch(pack):
+@pytest.mark.parametrize("resumed", [False, True])
+def test_observed_usage_real_cli_without_model_dispatch(pack, resumed):
+    from loopx.capabilities.benchmark_toolkit.trajectory_assessment_store import read_usage_turn
+
     scope, _, old, input_path, sources, _ = pack
-    (sources / 'usage.jsonl').write_text('\n'.join(_usage_fixture().splitlines()[:-1]) + '\n')
+    rows = [json.loads(line) for line in (_request_usage_fixture() if resumed else _usage_fixture()).splitlines()]
+    if resumed:
+        first_end = next(row for row in rows if row["payload"].get("type") == "task_complete")
+        first_end["payload"].update(type="turn_aborted", reason="SYNTHETIC-PRIVATE-ABORT-REASON")
+    else:
+        rows[-1]["payload"] = {"type": "turn_aborted", "reason": "SYNTHETIC-PRIVATE-ABORT-REASON"}
+        rows.append({"type": "event_msg", "timestamp": "2026-01-01T00:00:07Z",
+                     "payload": {"type": "token_count", "info": None}})
+    path = sources / "usage.jsonl"
+    path.write_text("\n".join(map(json.dumps, rows)) + "\n")
+    original = path.read_bytes()
     registry = sources.parent / 'registry-observed.json'
     registry.write_text('{"projects":{}}')
     prefix = [sys.executable, '-m', 'loopx.entrypoint', '--registry', str(registry),
@@ -1334,15 +1347,36 @@ def test_observed_usage_real_cli_without_model_dispatch(pack):
             '--input-json', str(input_path), '--source-root', str(sources), '--title', old['title'],
             '--usage-jsonl', 'usage.jsonl', '--usage-session-id', 'solver-session']
     default = subprocess.run([*prefix, *args], text=True, capture_output=True)
-    assert default.returncode != 0 and 'completed Turn' in default.stdout + default.stderr
+    reason = 'aborted Turn' if resumed else 'completed Turn'
+    assert default.returncode != 0 and reason in default.stdout + default.stderr
     observed = subprocess.run([*prefix, *args, '--include-incomplete-usage'], text=True, capture_output=True)
     assert observed.returncode == 0, observed.stdout + observed.stderr
     record = json.loads(observed.stdout)
     discovery = subprocess.run([*prefix, 'trajectory-packs', '--goal-id', scope['owner_goal_id']], text=True, capture_output=True)
     assert discovery.returncode == 0
     view = next(x for x in json.loads(discovery.stdout)['packs'] if x['pack_id'] == record['pack_id'])
-    assert view['solver_usage']['projection']['input_tokens'] == 140
-    assert view['solver_usage']['projection']['incomplete_turn']['usage']['input_tokens'] == 40
+    projection = view['solver_usage']['projection']
+    assert projection['input_tokens'] == 140 and projection['completed_turns'] == 1
+    if resumed:
+        assert projection['incomplete_turn'] is None
+        [interrupted] = projection['interrupted_turns']
+        assert interrupted['turn_id'] == 'a' and interrupted['usage']['input_tokens'] == 100
+        assert projection['request_coverage']['compaction_requests'] == 1
+        assert projection['request_coverage']['all_requests']['input_tokens'] == 160
+        evidence = read_usage_turn(**scope, pack_id=record['pack_id'],
+                                   analysis_key=record['analysis_key'], turn_id='a')
+        boundary = json.loads(evidence['text'])['events'][-1]['payload']
+        assert boundary == {'type': 'turn_aborted', 'turn_id': 'a'}
+        assert 'SYNTHETIC-PRIVATE-ABORT-REASON' not in evidence['text']
+    else:
+        assert projection['incomplete_turn']['usage']['input_tokens'] == 40
+        assert projection['incomplete_turn']['last_line'] == len(rows)
+        evidence = read_usage_turn(**scope, pack_id=record['pack_id'],
+                                   analysis_key=record['analysis_key'], turn_id='b')
+        events = json.loads(evidence['text'])['events']
+        assert [event['line'] for event in events] == [3, 5, 6]
+        assert 'SYNTHETIC-PRIVATE-ABORT-REASON' not in evidence['text']
+    assert path.read_bytes() == original
     assert inspect_pack(**scope, pack_id=record['pack_id'])['job'] is None
     assert not list(scope['runtime_root'].rglob('receipt.json'))
 
diff --git a/tests/control_plane_ts/trajectory_request_usage.test.ts b/tests/control_plane_ts/trajectory_request_usage.test.ts
index c71a92c855..4e1a532873 100644
--- a/tests/control_plane_ts/trajectory_request_usage.test.ts
+++ b/tests/control_plane_ts/trajectory_request_usage.test.ts
@@ -122,6 +122,45 @@ test('unfinished requests retain explicit partial admission; forks and malformed
   ]){const rows=fixture();mutate(rows);assert.throws(()=>run(rows,true));}
 });
 
+test('an identified interruption keeps observed requests across resume without completing the aborted Turn', () => {
+  const rows = fixture();
+  rows[6].payload = {type: 'turn_aborted', turn_id: 'a'};
+  assert.throws(() => run(rows), /aborted Turn/);
+  const resumed = run(rows, true);
+  assert.deepEqual(resumed.turns.map(turn => turn.turn_id), ['b']);
+  assert.equal(resumed.completed_turns, 1);
+  assert.equal(resumed.incomplete_turn, null);
+  assert.equal(resumed.input_tokens, 140);
+  assert.equal(resumed.turn_elapsed_seconds, 8);
+  assert.deepEqual(resumed.interrupted_turns, [{
+    turn_id: 'a', start: rows[1].timestamp, observed_through: rows[6].timestamp,
+    first_line: 2, last_line: 7, counter_observations: 1,
+    usage: {input_tokens: 100, cached_input_tokens: 60, uncached_input_tokens: 40, output_tokens: 10},
+  }]);
+  const coverage = resumed.request_coverage!;
+  assert.equal(coverage.status, 'verified');
+  if (coverage.status === 'verified') {
+    assert.equal(coverage.unique_requests, 3);
+    assert.equal(coverage.compaction_requests, 1);
+    assert.deepEqual(coverage.all_requests, tokens(160, 90, 22));
+  }
+  const wrong = structuredClone(rows);
+  wrong[6].payload.turn_id = 'b';
+  assert.throws(() => run(wrong, true), /unmatched aborted Turn/);
+  const anonymous = structuredClone(rows);
+  delete anonymous[6].payload.turn_id;
+  assert.throws(() => run(anonymous, true), /overlapping Turn/);
+  const duplicate = structuredClone(rows);
+  duplicate[7].payload.turn_id = 'a';
+  assert.throws(() => run(duplicate, true), /overlapping Turn/);
+  const interruptedOnly = run(renumber([...rows.slice(0, 7), {
+    ...event('token_count'), timestamp: rows[7].timestamp,
+  }]), true);
+  assert.equal(interruptedOnly.completed_turns, 0);
+  assert.equal(interruptedOnly.interrupted_turns?.length, 1);
+  assert.equal(interruptedOnly.end, rows[6].timestamp);
+});
+
 test('a host without request records is unknown, while verified zero compactions is an observed zero', () => {
   const es=fixture().filter(e=>e.type!=='compacted' && e.payload.response_id!=='compact');
   es.find(e=>e.payload.response_id==='two').payload.thread_token_usage=tokens(140,80,15);