huangruiteng · Nov 11, 2026 · Nov 11, 2026 · Nov 11, 2026
@@ -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/);

@@ -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) {

@@ -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):

@@ -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

@@ -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;

@@ -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();

@@ -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")

@@ -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",

@@ -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;