fix(parser): force full parse when a queued command masks the split check

The engine's cross-sync split detection compares only the first appended
message's ClaudeMessageID against the stored tail. Queued-command
attachments are written mid-stream between same-message.id assistant
chunks, so one falling inside a run that straddles a sync boundary
sorts ahead of the continuation head and hides it, appending the
stored partial response a second time. Detect the masking in
claudeParseSessionFrom and fall back to a replacing full parse.

Author: wesm

docs/internal/session-format-sources.md

@@ -70,7 +70,12 @@ Grok section and remove the explicit registry exception in the coverage test.
   cost field is consumed; Agentsview prices the tokens from its catalog.
 - **Agentsview:** `internal/parser/claude.go` and
   `internal/parser/claude_provider.go`; local observations and fixtures are
-  the implementation evidence for fields not documented upstream.
+  the implementation evidence for fields not documented upstream. Reverified
+  2026-07-22 against local CLI transcripts: `type=attachment` records with
+  `attachment.type=queued_command` are written mid-stream, in file order
+  between consecutive `assistant` records that share one `message.id`, so a
+  queued command can fall inside a streaming run that straddles an incremental
+  sync boundary.
 
 ## OpenClaude (`openclaude`)
 

internal/parser/claude.go

@@ -695,7 +695,9 @@ func claudeParseSessionFrom(
 	// happens in the full parser's post-merge uuid set, driving such
 	// files to linear parsing. Runs that straddle a sync boundary are
 	// detected by the engine's LastClaudeMessageID check on the first
-	// appended assistant message.
+	// appended assistant message; when a queued command would sort
+	// ahead of that head and mask the check, the parser itself falls
+	// back to a full parse (claudeQueuedCommandMasksSplitDetection).
 	if len(entries) > 1 {
 		entries = mergeClaudeAssistantMessageChunks(entries)
 	}
@@ -763,6 +765,16 @@ func claudeParseSessionFrom(
 	)
 	annotateSubagentSessions(msgs, subagentMap)
 	if len(queuedCommands) > 0 {
+		// The engine's cross-sync split detection compares only the
+		// FIRST returned message's ClaudeMessageID against the stored
+		// tail. A queued command sorting ahead of an assistant head
+		// would mask a same-message.id continuation and let the stored
+		// partial response be appended a second time instead of
+		// replaced — fall back to a full parse instead.
+		if claudeQueuedCommandMasksSplitDetection(msgs, queuedCommands) {
+			return nil, nil, time.Time{}, 0,
+				ErrClaudeIncrementalNeedsFullParse
+		}
 		msgs = mergeQueuedCommands(
 			msgs, queuedCommands, startOrdinal, queuedCommandMessage,
 		)
@@ -1487,6 +1499,33 @@ func mergeQueuedCommands(
 	return out
 }
 
+// claudeQueuedCommandMasksSplitDetection reports whether merging
+// queued commands would sort one ahead of a leading assistant message
+// that carries a provider message id. The engine's cross-sync split
+// detection (LastClaudeMessageID) inspects only the first appended
+// message, so a displaced assistant head would hide a same-message.id
+// continuation of the stored tail and duplicate the partial response.
+// Real CLI transcripts write queued_command attachments mid-stream,
+// between chunks of one response, so this masking is reachable
+// whenever the sync boundary falls inside such a run.
+func claudeQueuedCommandMasksSplitDetection(
+	msgs []ParsedMessage, queued []claudeQueuedCommand,
+) bool {
+	if len(msgs) == 0 {
+		return false
+	}
+	head := msgs[0]
+	if head.Role != RoleAssistant || head.ClaudeMessageID == "" {
+		return false
+	}
+	for _, qc := range queued {
+		if queuedBefore(qc, head) {
+			return true
+		}
+	}
+	return false
+}
+
 // queuedBefore reports whether a queued_command should sort before
 // a regular message. Zero timestamps on either side are treated
 // conservatively: a zero-timestamp message keeps its original

internal/parser/claude_parser_test.go

@@ -1486,6 +1486,151 @@ func TestParseClaudeSessionFrom_SameMessageIDRunMergesIncrementally(
 	assert.Equal(t, RoleUser, newMsgs[1].Role)
 }
 
+// A queued_command attachment can be written mid-stream, between
+// assistant chunks of one same-message.id response (observed in real
+// CLI transcripts). When the sync boundary falls inside that run, the
+// append window is [queued command, chunk, chunk] and the queued
+// command's earlier timestamp sorts it to position 0 of the returned
+// messages. The engine's cross-sync split detection compares only the
+// FIRST appended message's ClaudeMessageID against the stored tail, so
+// a masked head would append the continuation as a duplicate message.
+// The parser must fall back to a full parse instead.
+func TestParseClaudeSessionFrom_QueuedCommandBeforeContinuationFallsBack(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	chunk := func(uuid, parent, ts, text string) string {
+		return `{"type":"assistant","uuid":"` + uuid +
+			`","parentUuid":"` + parent +
+			`","timestamp":"` + ts +
+			`","message":{"id":"msg_split","content":[` +
+			`{"type":"text","text":"` + text + `"}]}}`
+	}
+
+	initial := testjsonl.JoinJSONL(
+		testjsonl.ClaudeUserJSON("hello", tsEarly),
+		chunk("a1", "u1", "2024-01-01T10:00:01Z", "Hel"),
+	)
+
+	parseFrom := func(
+		t *testing.T, appended string,
+	) ([]ParsedMessage, error) {
+		t.Helper()
+		path := createTestFile(
+			t, "inc-queued-boundary.jsonl", initial,
+		)
+		info, err := os.Stat(path)
+		require.NoError(t, err)
+		offset := info.Size()
+
+		f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644)
+		require.NoError(t, err)
+		_, err = f.WriteString(appended)
+		require.NoError(t, err)
+		require.NoError(t, f.Close())
+
+		msgs, _, _, _, perr := claudeParseSessionFrom(
+			path, offset, claudeIncrementalScan{
+				startOrdinal:      2,
+				lastEntryUUID:     "a1",
+				storedLinearParse: new(true),
+			},
+		)
+		return msgs, perr
+	}
+
+	t.Run("queued command masks continuation head", func(t *testing.T) {
+		t.Parallel()
+
+		appended := testjsonl.ClaudeQueuedCommandJSON(
+			"queued mid-stream", "2024-01-01T10:00:02Z",
+		) + "\n" +
+			chunk("a2", "a1", "2024-01-01T10:00:03Z", "Hello wo") + "\n" +
+			chunk("a3", "a2", "2024-01-01T10:00:04Z", "Hello world") + "\n"
+
+		_, perr := parseFrom(t, appended)
+		assert.ErrorIs(t, perr, ErrClaudeIncrementalNeedsFullParse,
+			"a queued command sorting ahead of a same-message.id "+
+				"continuation must force a full parse")
+	})
+
+	t.Run("full parse of the whole file keeps one merged message",
+		func(t *testing.T) {
+			t.Parallel()
+
+			content := initial + "\n" + testjsonl.JoinJSONL(
+				testjsonl.ClaudeQueuedCommandJSON(
+					"queued mid-stream", "2024-01-01T10:00:02Z",
+				),
+				chunk("a2", "a1", "2024-01-01T10:00:03Z", "Hello wo"),
+				chunk("a3", "a2", "2024-01-01T10:00:04Z", "Hello world"),
+			)
+			_, msgs := runClaudeParserTest(
+				t, "queued-boundary-full.jsonl", content,
+			)
+			require.Len(t, msgs, 3)
+			assert.Equal(t, RoleUser, msgs[0].Role)
+			assert.Equal(t, "queued_command", msgs[1].SourceSubtype)
+			assert.Equal(t, RoleAssistant, msgs[2].Role)
+			assert.Equal(t, "Hello world", msgs[2].Content,
+				"the run must collapse to one merged assistant message")
+			ids := 0
+			for _, m := range msgs {
+				if m.ClaudeMessageID == "msg_split" {
+					ids++
+				}
+			}
+			assert.Equal(t, 1, ids,
+				"msg_split must appear exactly once after a full parse")
+		})
+
+	t.Run("queued command after the run stays incremental",
+		func(t *testing.T) {
+			t.Parallel()
+
+			appended := chunk(
+				"a2", "a1", "2024-01-01T10:00:03Z", "Hello wo",
+			) + "\n" +
+				chunk("a3", "a2", "2024-01-01T10:00:04Z", "Hello world") +
+				"\n" + testjsonl.ClaudeQueuedCommandJSON(
+				"queued after", "2024-01-01T10:02:00Z",
+			) + "\n"
+
+			msgs, perr := parseFrom(t, appended)
+			require.NoError(t, perr)
+			require.Len(t, msgs, 2)
+			assert.Equal(t, RoleAssistant, msgs[0].Role)
+			assert.Equal(t, "msg_split", msgs[0].ClaudeMessageID,
+				"the continuation head must stay first so the engine's "+
+					"split check can see it")
+			assert.Equal(t, "queued_command", msgs[1].SourceSubtype)
+		})
+
+	t.Run("queued command before a user head stays incremental",
+		func(t *testing.T) {
+			t.Parallel()
+
+			appended := testjsonl.ClaudeQueuedCommandJSON(
+				"queued early", "2024-01-01T10:00:02Z",
+			) + "\n" +
+				`{"type":"user","uuid":"u2","parentUuid":"a1",` +
+				`"timestamp":"2024-01-01T10:00:03Z",` +
+				`"message":{"content":"next turn"}}` + "\n" +
+				`{"type":"assistant","uuid":"a2","parentUuid":"u2",` +
+				`"timestamp":"2024-01-01T10:00:04Z",` +
+				`"message":{"id":"msg_next","content":[` +
+				`{"type":"text","text":"fresh"}]}}` + "\n"
+
+			msgs, perr := parseFrom(t, appended)
+			require.NoError(t, perr)
+			require.Len(t, msgs, 3)
+			assert.Equal(t, "queued_command", msgs[0].SourceSubtype)
+			assert.Equal(t, "next turn", msgs[1].Content)
+			assert.Equal(t, "msg_next", msgs[2].ClaudeMessageID)
+		})
+}
+
 // An appended parentless entry adds a DAG root — the only property of
 // the stored linearity verdict an append can move toward
 // resolvability — so even a linear-bound session must fall back to a