From 8aaef2c6986e91ff3dc6bb33a1a799fbd53b933b Mon Sep 17 00:00:00 2001 From: Antisophy <293439221+Antisophy@users.noreply.github.com> Date: Sat, 15 Aug 2026 19:15:00 -0700 Subject: [PATCH] fix(history): treat mid-turn absorption as delivery, not withdrawal Recent Claude CLIs absorb queued messages into the running turn at a tool boundary instead of holding them for the next one: the entry is removed from the queue *after* delivery, and the delivery is recorded as a queued_command attachment rather than as a user line. Every remove was reported as consumed_as "removed", so messages the model demonstrably received were badged "not delivered", and the delivery itself was invisible because attachment records were never translated. Translate queued_command attachments into user messages and pair each one strictly with a remove. The pair's records land in either order, so an attachment against a still-queued entry is held until the entry's fate is known: a remove completes the absorption and releases it, a dequeue discards it (the echo carries the message in that flow, where recent CLIs also write the attachment as presentation), and one matching nothing renders nothing. An entry removed with no delivery record keeps the not-delivered outcome, so a genuine queue clear still reports honestly. Queue records are matched resiliently, because content is an unreliable key: the CLI omits it entirely for array-valued messages and texts can repeat. Matching prefers an exact text match, then empty-to-empty (an empty search names an array message, whose entry is stored empty too), and a remove finally settles the oldest unsettled entry, since the queue really did remove one. Echo pairing follows the same principle: dequeue records do not map enqueues one to one (the CLI can drain several entries under one dequeue, joining their texts into one echo), so the echo prefers the awaiting entry whose text it carries and a joined echo settles every queued entry whose text appears in it. Any entry left permanently unsettled would shift every later pairing behind by one, confirming each echo against the previous message's provisional and rendering every turn opener twice after reload. The delivery record is now the visible message, so it anchors checkpoints like any user line, and undo truncation takes its queue records with it, matched by the text they carry since the enqueue is not adjacent to the delivery. An absorbed delivery carries no file checkpoint (there is no user record for the CLI to checkpoint at), so its boundary offers conversation-only undo instead of a file revert the CLI would reject. --- source/cydo/agent/drivers/claude.d | 166 +++++++++ source/cydo/domain/tasks/model.d | 27 ++ source/cydo/server/app.d | 92 ++++- source/cydo/workflow/history/jsonl_store.d | 79 +++++ source/cydo/workflow/history/pipeline.d | 377 +++++++++++++++++++-- web/src/sessionReducer.test.ts | 74 ++++ 6 files changed, 786 insertions(+), 29 deletions(-) diff --git a/source/cydo/agent/drivers/claude.d b/source/cydo/agent/drivers/claude.d index 8445a530..59a8b8cd 100644 --- a/source/cydo/agent/drivers/claude.d +++ b/source/cydo/agent/drivers/claude.d @@ -446,6 +446,12 @@ class ClaudeCodeAgent : Agent continue; } bool isUser = line.canFind(`"type":"user"`); + // A mid-turn absorption records its delivery as a queued_command + // attachment rather than a user line, and that record is the + // visible message, so it anchors checkpoints like any other one. + if (!isUser && line.canFind(`"type":"attachment"`) + && line.canFind(`"queued_command"`)) + isUser = true; if (!isUser && !line.canFind(`"type":"assistant"`)) continue; enum prefix = `"uuid":"`; @@ -2174,6 +2180,9 @@ private TranslatedEvent[] translateClaudeHistoryEvent(string rawLine) case "user": result = normalizeUserHistory(rawLine); break; + case "attachment": + result = translateQueuedCommandAttachment(rawLine); + break; case "stream_event": return []; // not stored in JSONL history default: @@ -2484,6 +2493,163 @@ private TranslatedEvent[] normalizeUserHistory(string rawLine) return events; } +/// Translate a queued_command attachment record to a user message. +/// +/// Recent Claude CLIs absorb queued messages into the running turn at a tool +/// boundary instead of holding them for the next turn: it removes each entry +/// from the queue and records the delivery as an attachment rather than as a +/// user line, so this record is the only canonical fact for a message the +/// model demonstrably received. Its uuid is the one the CLI replays on stdout +/// for the same delivery, so live and replayed histories agree on identity. +/// +/// Older versions dequeued the whole queue at the turn boundary and wrote one +/// joined user line; those histories carry no attachment records and are +/// unaffected. +private TranslatedEvent[] translateQueuedCommandAttachment(string rawLine) +{ + import cydo.protocol : ContentBlock, ItemStartedEvent; + + @JSONPartial static struct Attachment + { + string type; + @JSONOptional JSONFragment prompt; + @JSONOptional string commandMode; + } + @JSONPartial static struct Record + { + Attachment attachment; + @JSONOptional string uuid; + @JSONOptional string parent_tool_use_id; + @JSONOptional bool isSidechain; + } + + Record raw; + try + raw = jsonParse!Record(rawLine); + catch (Exception e) + { tracef("translateQueuedCommandAttachment: parse error: %s", e.msg); return []; } + + if (raw.attachment.type != "queued_command") + return []; + + auto promptJson = raw.attachment.prompt.json; + if (promptJson is null || promptJson.length == 0) + return []; + + ContentBlock[] blocks; + if (promptJson[0] == '"') + { + string text; + try text = jsonParse!string(promptJson); + catch (Exception) { return []; } + ContentBlock cb; + cb.type = "text"; + cb.text = text; + blocks ~= cb; + } + else if (promptJson[0] == '[') + { + @JSONPartial static struct ImageSource + { + @JSONOptional string data; + @JSONOptional string media_type; + } + @JSONPartial static struct PromptBlock + { + string type; + @JSONOptional string text; + @JSONOptional ImageSource source; + } + PromptBlock[] items; + try items = jsonParse!(PromptBlock[])(promptJson); + catch (Exception e) + { tracef("translateQueuedCommandAttachment: prompt parse error: %s", e.msg); return []; } + foreach (ref item; items) + { + ContentBlock cb; + if (item.type == "text") + { + cb.type = "text"; + cb.text = item.text; + } + else if (item.type == "image") + { + cb.type = "image"; + cb.data = item.source.data; + cb.media_type = item.source.media_type; + } + else + continue; + blocks ~= cb; + } + } + if (blocks.length == 0) + return []; + + ItemStartedEvent ev; + ev.item_id = "cc-queued-command"; + ev.item_type = "user_message"; + ev.content = blocks; + ev.uuid = raw.uuid; + ev.parent_tool_use_id = raw.parent_tool_use_id; + ev.is_sidechain = raw.isSidechain; + return [TranslatedEvent(toJson(ev), rawLine)]; +} + +/// Prompt text carried by a queued_command attachment record, or null when the +/// line is not one. Used to pair a delivery with the queue entry it consumed. +package(cydo) string queuedCommandAttachmentPrompt(string rawLine) +{ + import std.algorithm : canFind; + + if (!rawLine.canFind(`"queued_command"`)) + return null; + auto events = translateQueuedCommandAttachment(rawLine); + if (events.length == 0) + return null; + + @JSONPartial static struct ContentBlockProbe + { + @JSONOptional string type; + @JSONOptional string text; + } + @JSONPartial static struct Probe + { + @JSONOptional ContentBlockProbe[] content; + @JSONOptional string uuid; + } + Probe probe; + try + probe = jsonParse!Probe(events[0].translated); + catch (Exception) + return null; + foreach (ref block; probe.content) + if (block.type == "text") + return block.text; + return null; +} + +/// Uuid of a queued_command attachment record, or null when the line is not +/// one. This is the identity the delivered message carries. +package(cydo) string queuedCommandAttachmentUuid(string rawLine) +{ + import std.algorithm : canFind; + + if (!rawLine.canFind(`"queued_command"`)) + return null; + @JSONPartial static struct Probe + { + @JSONOptional string type; + @JSONOptional string uuid; + } + Probe probe; + try + probe = jsonParse!Probe(rawLine); + catch (Exception) + return null; + return probe.type == "attachment" ? probe.uuid : null; +} + /// Translate a Claude stream-json event to the agent-agnostic protocol. /// Returns TranslatedEvent.init for events that should be consumed (not forwarded). private TranslatedEvent translateClaudeEvent(string rawLine, string agentName) diff --git a/source/cydo/domain/tasks/model.d b/source/cydo/domain/tasks/model.d index 4caacf27..c44e3ab5 100644 --- a/source/cydo/domain/tasks/model.d +++ b/source/cydo/domain/tasks/model.d @@ -636,21 +636,48 @@ struct TaskData // nonces awaiting their enqueue record for identity linking. --- string[] queueTailQueuedUuids; string[] queueTailQueuedNonces; + // Enqueued text, kept so a mid-turn delivery can be matched back to the + // entry it consumed: the queued_command attachment that records the + // delivery carries the text, not the queue position. + string[] queueTailQueuedContents; + // Native uuid of a queued_command attachment seen while its entry was + // still queued; only a following remove proves it was the absorbed + // delivery (a dequeue means the echo carries the message). + string[] queueTailQueuedHeldUuids; string[] queueTailAwaitingUuids; string[] queueTailAwaitingNonces; string[] sentNonceFifo; + // Entries removed from the queue whose delivery record has not arrived yet. + // A mid-turn absorption writes the remove and the delivery in either order, + // so a removal is only conclusive once the surrounding lines are seen. + string[] queueTailRemovedUuids; + string[] queueTailRemovedNonces; + string[] queueTailRemovedContents; invariant (queueTailQueuedUuids.length == queueTailQueuedNonces.length, "queue tail queued arrays length mismatch"); + invariant (queueTailQueuedUuids.length == queueTailQueuedContents.length, + "queue tail queued content array length mismatch"); + invariant (queueTailQueuedUuids.length == queueTailQueuedHeldUuids.length, + "queue tail held uuid array length mismatch"); invariant (queueTailAwaitingUuids.length == queueTailAwaitingNonces.length, "queue tail awaiting arrays length mismatch"); + invariant (queueTailRemovedUuids.length == queueTailRemovedNonces.length, + "queue tail removed arrays length mismatch"); + invariant (queueTailRemovedUuids.length == queueTailRemovedContents.length, + "queue tail removed content array length mismatch"); void clearQueueTailState() { queueTailQueuedUuids = null; queueTailQueuedNonces = null; + queueTailQueuedContents = null; + queueTailQueuedHeldUuids = null; queueTailAwaitingUuids = null; queueTailAwaitingNonces = null; + queueTailRemovedUuids = null; + queueTailRemovedNonces = null; + queueTailRemovedContents = null; sentNonceFifo = null; } diff --git a/source/cydo/server/app.d b/source/cydo/server/app.d index 812e125b..55435ee7 100644 --- a/source/cydo/server/app.d +++ b/source/cydo/server/app.d @@ -2646,7 +2646,9 @@ class App /// message was withdrawn without ever being consumed. private void onTailedJsonlLine(int tid, string line, int lineNum) { - import std.algorithm : startsWith; + import std.algorithm : remove, startsWith; + import cydo.agent.drivers.claude : queuedCommandAttachmentPrompt, + queuedCommandAttachmentUuid; auto td = tid in tasks; if (td is null) @@ -2676,30 +2678,77 @@ class App } td.queueTailQueuedUuids ~= format!"enqueue-%d"(lineNum); td.queueTailQueuedNonces ~= nonce; + td.queueTailQueuedContents ~= op.content; + td.queueTailQueuedHeldUuids ~= null; } else if (op.operation == "dequeue") { + // the echo carries this message; a held attachment was + // presentation-only, not the delivery if (td.queueTailQueuedUuids.length > 0) { td.queueTailAwaitingUuids ~= td.queueTailQueuedUuids[0]; td.queueTailAwaitingNonces ~= td.queueTailQueuedNonces[0]; - td.queueTailQueuedUuids = td.queueTailQueuedUuids[1 .. $]; - td.queueTailQueuedNonces = td.queueTailQueuedNonces[1 .. $]; + dropQueueTailEntry(td, 0); } } else if (op.operation == "remove") { - if (td.queueTailQueuedUuids.length > 0) + // Mid-turn absorption removes an entry after delivering it, + // while a queue clear removes one that was never delivered. + // Only the queued_command attachment tells them apart, and it + // can arrive on either side of this record, so hold the entry + // until the delivery resolves it. + auto idx = queueTailIndexOfContent(td, op.content); + if (idx >= 0) { - emitUserMessageConsumed(tid, td.queueTailQueuedUuids[0], - "removed", td.queueTailQueuedNonces[0]); - td.queueTailQueuedUuids = td.queueTailQueuedUuids[1 .. $]; - td.queueTailQueuedNonces = td.queueTailQueuedNonces[1 .. $]; + if (td.queueTailQueuedHeldUuids[idx].length > 0) + { + // the attachment arrived first; the remove completes + // the absorption pair + emitUserMessageConsumed(tid, td.queueTailQueuedUuids[idx], + "steering", td.queueTailQueuedNonces[idx], + td.queueTailQueuedHeldUuids[idx]); + dropQueueTailEntry(td, idx); + } + else + { + td.queueTailRemovedUuids ~= td.queueTailQueuedUuids[idx]; + td.queueTailRemovedNonces ~= td.queueTailQueuedNonces[idx]; + td.queueTailRemovedContents ~= td.queueTailQueuedContents[idx]; + dropQueueTailEntry(td, idx); + } } } return; } + if (auto prompt = queuedCommandAttachmentPrompt(line)) + { + // The delivery record: the message reached the model inside the + // running turn. Confirm the provisional bubble and point at the + // canonical message, which the CLI replays on stdout live and which + // this record itself becomes on reload. + auto nativeUuid = queuedCommandAttachmentUuid(line); + foreach (i, content; td.queueTailRemovedContents) + if (content == prompt) + { + emitUserMessageConsumed(tid, td.queueTailRemovedUuids[i], + "steering", td.queueTailRemovedNonces[i], nativeUuid); + td.queueTailRemovedUuids = td.queueTailRemovedUuids.remove(i); + td.queueTailRemovedNonces = td.queueTailRemovedNonces.remove(i); + td.queueTailRemovedContents = td.queueTailRemovedContents.remove(i); + return; + } + auto idx = queueTailIndexOfContent(td, prompt); + if (idx >= 0) + // still queued: only a following remove proves this attachment + // was the absorbed delivery (a dequeue means the echo carries + // the message), so hold it until the entry's fate is known + td.queueTailQueuedHeldUuids[idx] = nativeUuid; + return; + } + if (td.queueTailAwaitingUuids.length == 0) return; auto ta = tryAgentForTask(tid); @@ -2735,6 +2784,33 @@ class App td.queueTailAwaitingNonces = td.queueTailAwaitingNonces[1 .. $]; } + /// Index of the queued entry a record names, or -1. Content is an + /// unreliable key (omitted for array-valued messages, repeatable), so the + /// search prefers an exact match, then empty-to-empty (an empty search + /// names an array message, stored empty too), then the oldest entry, so a + /// mismatch can never leave an entry permanently unsettled. + private static ptrdiff_t queueTailIndexOfContent(TaskData* td, string content) + { + if (content.length > 0) + foreach (i, queued; td.queueTailQueuedContents) + if (queued == content) + return i; + foreach (i, queued; td.queueTailQueuedContents) + if (queued.length == 0) + return i; + return td.queueTailQueuedUuids.length > 0 ? 0 : -1; + } + + /// Drop one entry from the parallel queue-tail arrays. + private static void dropQueueTailEntry(TaskData* td, ptrdiff_t idx) + { + import std.algorithm : remove; + td.queueTailQueuedUuids = td.queueTailQueuedUuids.remove(idx); + td.queueTailQueuedNonces = td.queueTailQueuedNonces.remove(idx); + td.queueTailQueuedContents = td.queueTailQueuedContents.remove(idx); + td.queueTailQueuedHeldUuids = td.queueTailQueuedHeldUuids.remove(idx); + } + /// Append a user_message/consumed confirmation to task history and /// broadcast it to subscribed clients. private void emitUserMessageConsumed(int tid, string uuid, string consumedAs, diff --git a/source/cydo/workflow/history/jsonl_store.d b/source/cydo/workflow/history/jsonl_store.d index 6d3cce4d..3e1426ba 100644 --- a/source/cydo/workflow/history/jsonl_store.d +++ b/source/cydo/workflow/history/jsonl_store.d @@ -8,6 +8,7 @@ import std.string : representation; import ae.sys.data : Data; import ae.utils.json : JSONFragment, toJson; +import cydo.agent.drivers.claude : queuedCommandAttachmentPrompt; import cydo.protocol : TaskEventEnvelope, TranslatedEvent; import cydo.agent.contract : InterruptedToolCallRepair; import cydo.domain.storage.persistence : LoadedHistory, Persistence, createForkTask, noSourceLine; @@ -452,6 +453,25 @@ int truncateJsonl(string jsonlPath, string afterForkId, kept = kept[0 .. $ - 1]; removedCount++; } + // A mid-turn absorption's enqueue does not sit next to its + // delivery: the turn's assistant and tool_result lines run + // between them. Drop that entry's queue records wherever they + // are, keyed by the text they carry, or the undone message + // comes back as a queued one. + if (auto prompt = queuedCommandAttachmentPrompt(line)) + { + size_t write_ = 0; + foreach (keptLine; kept) + { + if (isQueueOperationForContent(keptLine, prompt)) + { + removedCount++; + continue; + } + kept[write_++] = keptLine; + } + kept = kept[0 .. write_]; + } continue; } } @@ -589,6 +609,48 @@ unittest assert(countLinesAfterForkId(jsonlPath, "enqueue-1", matchEnqueue, countForkable) == 2); } +// Undoing a mid-turn absorption must take its queue records with it. The +// enqueue sits far from the delivery (the turn's own output runs between +// them), so the adjacent-run-up sweep cannot reach it, and a survivor would +// resurrect the undone message as a still-queued one on the next load. +unittest +{ + import std.array : join; + import std.algorithm : canFind; + import std.file : mkdirRecurse, readText, rmdirRecurse, write; + import std.path : buildPath; + + auto dir = buildPath("/tmp", "cydo-persist-truncate-absorbed"); + mkdirRecurse(dir); + scope(exit) rmdirRecurse(dir); + + auto jsonlPath = buildPath(dir, "events.jsonl"); + write(jsonlPath, [ + `{"type":"user","uuid":"u1","message":{"role":"user","content":"start the turn"}}`, + `{"type":"queue-operation","operation":"enqueue","content":"absorbed text"}`, + `{"type":"assistant","uuid":"a1","message":{"content":[{"type":"text","text":"working"}]}}`, + `{"type":"user","uuid":"tr1","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t1","content":"ok"}]}}`, + `{"type":"queue-operation","operation":"remove","content":"absorbed text"}`, + `{"type":"attachment","uuid":"native-1","attachment":{"type":"queued_command","prompt":"absorbed text","commandMode":"prompt"}}`, + `{"type":"assistant","uuid":"a2","message":{"content":[{"type":"text","text":"answered"}]}}`, + ].join("\n") ~ "\n"); + + auto removed = truncateJsonl(jsonlPath, "native-1", + (string line, int lineNum, string forkId) => line.canFind(`"uuid":"` ~ forkId ~ `"`), + true); + assert(removed > 0, "the delivery must be found"); + + auto text = readText(jsonlPath); + assert(!text.canFind("absorbed text"), + "no trace of the undone message may remain, queue records included"); + assert(!text.canFind("native-1"), "the delivery record itself is gone"); + // everything before the message is untouched + assert(text.canFind(`"uuid":"u1"`) && text.canFind(`"uuid":"a1"`) + && text.canFind(`"uuid":"tr1"`), "prior turn content must survive"); + // and everything after it is truncated, as undo means "from here on" + assert(!text.canFind(`"uuid":"a2"`), "content after the message is truncated"); +} + unittest { import std.array : join; @@ -655,6 +717,23 @@ private bool isNeutralOrQueueOp(string line) || line.canFind(`"progress"`); } +/// Whether the line is a queue-operation record for this exact message text. +/// Queue records identify their entry by the text it carries, which is what +/// links an absorbed message's enqueue and remove back to its delivery. +private bool isQueueOperationForContent(string line, string content) +{ + import std.algorithm : canFind; + import ae.utils.json : jsonParse, JSONOptional, JSONPartial; + + if (content.length == 0 || !line.canFind(`"queue-operation"`)) + return false; + @JSONPartial static struct Probe { @JSONOptional string content; } + try + return jsonParse!Probe(line).content == content; + catch (Exception) + return false; +} + /// Extract a string field value from a JSON line by prefix scanning. private string extractJsonField(string line, string prefix) { diff --git a/source/cydo/workflow/history/pipeline.d b/source/cydo/workflow/history/pipeline.d index b4e1d6e4..b1c12158 100644 --- a/source/cydo/workflow/history/pipeline.d +++ b/source/cydo/workflow/history/pipeline.d @@ -16,6 +16,8 @@ import ae.utils.json : JSONFragment, JSONOptional, JSONPartial, jsonParse, toJso import ae.utils.time.types : AbsTime; import cydo.agent.contract : Agent, PersistedHistoryBoundaryKind; +import cydo.agent.drivers.claude : queuedCommandAttachmentPrompt, + queuedCommandAttachmentUuid; import cydo.protocol : ContentBlock, ItemStartedEvent, TaskEventEnvelope, TaskHistoryBoundaryReplacedEnvelope, TaskEventSeqEnvelope, TranslatedEvent, UnconfirmedUserEventEnvelope, UserMessageConsumedEvent, HistoryBoundary, @@ -130,8 +132,67 @@ class HistoryEventPipeline // and replaced by a user_message/consumed confirmation carrying the // CLI's own steering classification. Entries still queued at EOF stay // pending — a killed session's unconsumed steer remains visible. - string[] queuedUuids; - string[] awaitingEchoUuids; + // + // Mid-turn absorption (recent Claude CLIs) takes a third path: the entry + // is removed from the queue and delivered inside the running turn, with + // a queued_command attachment as the only record of the delivery. The + // attachment resolves the entry the same way an echo does; a remove + // with no attachment is a genuine drop and is settled at EOF. + struct QueueEntry + { + string uuid; + string content; + bool delivered; + bool removed; + // a queued_command attachment seen while the entry was still + // queued: only a following remove proves it was the absorbed + // delivery (a dequeue means the echo carries the message and the + // attachment is presentation-only) + TranslatedEvent[] heldDelivery; + string heldNativeUuid; + } + QueueEntry[] queueEntries; + static struct AwaitingEcho { string uuid; string content; } + AwaitingEcho[] awaitingEchoes; + /// Entry a remove settles, or -1. Content identifies the entry when it + /// matches, but content is an unreliable key: the CLI omits it for + /// array-valued messages, and texts can repeat. So the search prefers + /// an exact text match, then empty-to-empty (an empty search names an + /// array message, whose entry is also stored empty), then the oldest + /// entry not yet removed: the queue really did remove one, and leaving + /// an entry permanently unsettled shifts every later pairing behind by + /// one. + ptrdiff_t findEntryForRemove(string content) + { + if (content.length > 0) + foreach (i, ref entry; queueEntries) + if (!entry.removed && !entry.delivered && entry.content == content) + return i; + foreach (i, ref entry; queueEntries) + if (!entry.removed && !entry.delivered && entry.content.length == 0) + return i; + foreach (i, ref entry; queueEntries) + if (!entry.removed && !entry.delivered) + return i; + return -1; + } + + /// Entry a delivery record resolves, or -1. Same keying rules as the + /// remove search, but removed entries stay eligible (the remove lands + /// on either side of the attachment) and there is no oldest-entry + /// fallback: an attachment that matches nothing belongs to a message + /// the echo carries. + ptrdiff_t findEntryForDelivery(string content) + { + if (content.length > 0) + foreach (i, ref entry; queueEntries) + if (!entry.delivered && entry.content == content) + return i; + foreach (i, ref entry; queueEntries) + if (!entry.delivered && entry.content.length == 0) + return i; + return -1; + } auto stripTransientStatus = (TranslatedEvent[] events) { foreach (ref e; events) e.translated = host_.injectAgentNameIntoSessionInit(e.translated, td.agentName); @@ -167,7 +228,7 @@ class HistoryEventPipeline { hasQueueOps = true; auto enqueueUuid = format!"enqueue-%d"(lineNum); - queuedUuids ~= enqueueUuid; + queueEntries ~= QueueEntry(enqueueUuid, op.content); auto synEv = buildSyntheticUserEvent(op.content, false, true); synEv.uuid = enqueueUuid; return stripTransientStatus([TranslatedEvent( @@ -176,28 +237,80 @@ class HistoryEventPipeline } else if (op.operation == "dequeue") { - if (queuedUuids.length > 0) + // the echo will carry this message; a held attachment + // was presentation-only, not the delivery + foreach (i, ref entry; queueEntries) { - awaitingEchoUuids ~= queuedUuids[0]; - queuedUuids = queuedUuids[1 .. $]; + if (entry.removed || entry.delivered) + continue; + awaitingEchoes ~= AwaitingEcho(entry.uuid, entry.content); + queueEntries = queueEntries[0 .. i] + ~ queueEntries[i + 1 .. $]; + break; } return []; } else if (op.operation == "remove") { - TranslatedEvent[] result; - if (queuedUuids.length > 0) + // A remove is ambiguous on its own: mid-turn absorption + // removes an entry *after* delivering it, while a queue + // clear removes one that was never delivered. Only the + // queued_command attachment tells them apart, and it can + // land on either side of the remove, so the pair settles + // whenever it completes (or EOF settles a lone remove). + auto idx = findEntryForRemove(op.content); + if (idx < 0) + return []; + if (queueEntries[idx].heldDelivery.length > 0) { - result ~= TranslatedEvent(toJson(UserMessageConsumedEvent( - uuid: queuedUuids[0], consumed_as: "removed")), - null, AbsTime.init, lineNum); - queuedUuids = queuedUuids[1 .. $]; + auto entry = queueEntries[idx]; + queueEntries = queueEntries[0 .. idx] + ~ queueEntries[idx + 1 .. $]; + auto consumed = UserMessageConsumedEvent( + uuid: entry.uuid, + consumed_as: "steering", + native_uuid: entry.heldNativeUuid); + return stripTransientStatus( + [TranslatedEvent(toJson(consumed), + null, AbsTime.init, lineNum)] + ~ entry.heldDelivery); } - return stripTransientStatus(result); + queueEntries[idx].removed = true; + return []; } return []; } - if (awaitingEchoUuids.length > 0) + if (auto prompt = queuedCommandAttachmentPrompt(line)) + { + // A queued_command attachment is the delivery record only + // for a message the queue *removed* (mid-turn absorption); + // in the dequeue flow the echo carries the message and the + // attachment is presentation-only. The remove lands on + // either side of the attachment, so an attachment against a + // still-queued entry is held until the entry's fate is + // known; one that matches nothing (its message went out via + // dequeue, or it belongs to no tracked send) renders + // nothing. + auto idx = findEntryForDelivery(prompt); + if (idx < 0) + return []; + auto ts = ta.translateHistoryLine(line, lineNum); + if (!queueEntries[idx].removed) + { + queueEntries[idx].heldDelivery = ts; + queueEntries[idx].heldNativeUuid = + queuedCommandAttachmentUuid(line); + return []; + } + auto consumed = UserMessageConsumedEvent( + uuid: queueEntries[idx].uuid, + consumed_as: "steering", + native_uuid: queuedCommandAttachmentUuid(line)); + queueEntries = queueEntries[0 .. idx] ~ queueEntries[idx + 1 .. $]; + return stripTransientStatus([TranslatedEvent(toJson(consumed), + null, AbsTime.init, lineNum)] ~ ts); + } + if (awaitingEchoes.length > 0) { if (ta.isUserMessageLine(line)) { @@ -222,9 +335,48 @@ class HistoryEventPipeline // (anchors, checkpoints and truncation semantics attach // to it), preceded by the confirmation that tells the // UI to drop the provisional enqueue-emitted bubble. - auto enqueueUuid = awaitingEchoUuids[0]; - awaitingEchoUuids = awaitingEchoUuids[1 .. $]; auto ev = jsonParse!ItemStartedEvent(ts[0].translated); + // Prefer the awaiting entry whose text this echo carries: + // dequeue records do not map enqueues one to one (the CLI + // can drain several entries under one dequeue, joining + // their texts into one echo), so blind FIFO popping can + // pair the echo with a stale entry and shift every later + // pairing behind by one. + auto echoText = extractContentText(ev.content); + size_t pick = 0; + foreach (pi, ref waiting; awaitingEchoes) + if (waiting.content.length > 0 + && echoText.canFind(waiting.content)) + { + pick = pi; + break; + } + auto enqueueUuid = awaitingEchoes[pick].uuid; + awaitingEchoes = awaitingEchoes[0 .. pick] + ~ awaitingEchoes[pick + 1 .. $]; + TranslatedEvent[] joined; + // a joined echo consumed queued entries the dequeue records + // never covered; settle every entry whose text it carries so + // none linger as zombies shifting later pairings + for (size_t qi = 0; qi < queueEntries.length;) + { + if (!queueEntries[qi].removed && !queueEntries[qi].delivered + && queueEntries[qi].content.length > 0 + && queueEntries[qi].content != echoText + && echoText.canFind(queueEntries[qi].content)) + { + joined ~= TranslatedEvent(toJson( + UserMessageConsumedEvent( + uuid: queueEntries[qi].uuid, + consumed_as: "turn_start", + native_uuid: ev.uuid)), + null, AbsTime.init, lineNum); + queueEntries = queueEntries[0 .. qi] + ~ queueEntries[qi + 1 .. $]; + continue; + } + qi++; + } // Persisted echo lines carry no steering flag (it is a // live-stdout-only field); mid-turn consumption is // classified via the assistant-fallback branch below. @@ -233,8 +385,9 @@ class HistoryEventPipeline consumed_as: "turn_start", native_uuid: ev.uuid.length > 0 ? ev.uuid : enqueueUuid); return stripTransientStatus([ - TranslatedEvent(toJson(consumed), null, AbsTime.init, lineNum), - TranslatedEvent(toJson(ev), ts[0].raw)] ~ ts[1 .. $]); + TranslatedEvent(toJson(consumed), null, AbsTime.init, lineNum)] + ~ joined + ~ [TranslatedEvent(toJson(ev), ts[0].raw)] ~ ts[1 .. $]); } // not the echo (tool_result, or empty translation): pass // through unchanged and stay awaiting the real echo @@ -246,8 +399,8 @@ class HistoryEventPipeline // dequeued message was consumed; turn openers always echo // before assistant output, so classify as steering. auto consumed = UserMessageConsumedEvent( - uuid: awaitingEchoUuids[0], consumed_as: "steering"); - awaitingEchoUuids = awaitingEchoUuids[1 .. $]; + uuid: awaitingEchoes[0].uuid, consumed_as: "steering"); + awaitingEchoes = awaitingEchoes[1 .. $]; auto ts = ta.translateHistoryLine(line, lineNum); return stripTransientStatus([TranslatedEvent(toJson(consumed), null, AbsTime.init, lineNum)] ~ ts); @@ -264,6 +417,21 @@ class HistoryEventPipeline infof("Loaded history for task %d (%d events, %d ms)", tid, td.history.length, sw.peek.total!"msecs"); + // Removes whose delivery record never arrived: the message left the + // queue without being consumed (a queue clear), so its bubble keeps the + // not-delivered presentation. Settled here because a delivery may + // follow its remove by several lines, so no earlier point can tell. + foreach (ref entry; queueEntries) + { + if (!entry.removed || entry.delivered) + continue; + import std.datetime : Clock; + td.history.appendLive(Data( + toJson(TaskEventEnvelope(tid, Clock.currStdTime, + JSONFragment(toJson(UserMessageConsumedEvent( + uuid: entry.uuid, consumed_as: "removed"))))).representation), null); + } + if (orphan) appendTaskDiagnostic(tid, "Failed to load session history", buildOrphanAgentBody(td.agentName, @@ -630,6 +798,7 @@ private: @JSONOptional bool is_steering; @JSONOptional bool pending; @JSONOptional string uuid; + @JSONOptional string item_id; } UserAnchorProbe probe; @@ -645,8 +814,12 @@ private: auto uuid = probe.uuid; auto isEnqueue = uuid.length > "enqueue-".length && uuid.startsWith("enqueue-"); + // a mid-turn absorption's delivery record is an attachment, not a + // user message, so the CLI has no file checkpoint at it; the + // boundary stays undoable, conversation-only + auto isAbsorbedDelivery = probe.item_id == "cc-queued-command"; string checkpointUuid; - if (!isEnqueue && uuid.length > 0) + if (!isEnqueue && !isAbsorbedDelivery && uuid.length > 0) checkpointUuid = uuid; else if (rawLine.length > 0) { @@ -1365,6 +1538,168 @@ unittest assert(replayedDiagnostic.severity == "error"); } +// Mid-turn absorption (recent Claude CLIs): queued messages are delivered inside +// the running turn and recorded as queued_command attachments, with the queue +// remove following as a receipt. The delivery must resolve the provisional +// bubble rather than mark it undelivered, and the attachment order relative to +// its remove varies, so both orders are exercised. A remove with no delivery +// keeps the not-delivered outcome. +unittest +{ + import std.algorithm : canFind; + import std.array : join; + import std.file : exists, getSize, mkdirRecurse, rmdirRecurse, write; + import std.path : buildPath, dirName; + import std.process : environment; + import cydo.agent.drivers.claude : ClaudeCodeAgent; + import cydo.domain.tasks.model : Watermark; + import cydo.runtime.launch.types : NativeHistoryProfile; + import cydo.workflow.history.native_history : HistoryAccess; + + auto dir = buildPath("/tmp", "cydo-history-midturn-absorption"); + if (exists(dir)) + rmdirRecurse(dir); + mkdirRecurse(dir); + scope(exit) rmdirRecurse(dir); + + auto projectPath = buildPath(dir, "project"); + mkdirRecurse(projectPath); + + auto oldConfigDir = environment.get("CLAUDE_CONFIG_DIR"); + environment["CLAUDE_CONFIG_DIR"] = buildPath(dir, "claude"); + scope(exit) + { + if (oldConfigDir is null) + environment.remove("CLAUDE_CONFIG_DIR"); + else + environment["CLAUDE_CONFIG_DIR"] = oldConfigDir; + } + + enum tid = 1; + auto td = TaskData(tid, "local", projectPath); + td.agentName = "claude"; + td.agentSessionId = "S"; + td.worktreeTid = 0; + + Agent agent = new ClaudeCodeAgent(); + auto profile = NativeHistoryProfile(agent.driver, buildPath(dir, "claude")); + auto jsonlPath = agent.historyPath(td.agentSessionId, projectPath, profile); + mkdirRecurse(dirName(jsonlPath)); + // first message: the delivery follows its remove; second: it precedes; + // third is removed by a queue clear and never delivered + auto jsonl = [ + `{"type":"queue-operation","operation":"enqueue","timestamp":"2026-08-05T06:00:00Z","sessionId":"S","content":"first"}`, + `{"type":"queue-operation","operation":"remove","timestamp":"2026-08-05T06:00:01Z","sessionId":"S","content":"first"}`, + `{"type":"attachment","uuid":"native-first","timestamp":"2026-08-05T06:00:01Z","attachment":{"type":"queued_command","prompt":"first","commandMode":"prompt"}}`, + `{"type":"queue-operation","operation":"enqueue","timestamp":"2026-08-05T06:00:02Z","sessionId":"S","content":"second"}`, + `{"type":"attachment","uuid":"native-second","timestamp":"2026-08-05T06:00:03Z","attachment":{"type":"queued_command","prompt":"second","commandMode":"prompt"}}`, + `{"type":"queue-operation","operation":"remove","timestamp":"2026-08-05T06:00:03Z","sessionId":"S","content":"second"}`, + `{"type":"queue-operation","operation":"enqueue","timestamp":"2026-08-05T06:00:04Z","sessionId":"S","content":"dropped"}`, + `{"type":"queue-operation","operation":"remove","timestamp":"2026-08-05T06:00:05Z","sessionId":"S","content":"dropped"}`, + // dequeue flow: the CLI writes the attachment too, but the echo is the + // message; the attachment must render nothing + `{"type":"queue-operation","operation":"enqueue","timestamp":"2026-08-05T06:00:06Z","sessionId":"S","content":"echoed"}`, + `{"type":"attachment","uuid":"native-echoed","timestamp":"2026-08-05T06:00:06Z","attachment":{"type":"queued_command","prompt":"echoed","commandMode":"prompt"}}`, + `{"type":"queue-operation","operation":"dequeue","timestamp":"2026-08-05T06:00:07Z","sessionId":"S"}`, + `{"type":"user","uuid":"echo-user-1","message":{"role":"user","content":"echoed"}}`, + // an array-valued message: the CLI omits the enqueue and remove + // content, and only the attachment carries the text; the empty-to-empty + // pairing must deliver it rather than strand the entry, and the strand + // must not shift the following turn opener's pairing + `{"type":"queue-operation","operation":"enqueue","timestamp":"2026-08-05T06:00:08Z","sessionId":"S"}`, + `{"type":"queue-operation","operation":"remove","timestamp":"2026-08-05T06:00:09Z","sessionId":"S"}`, + `{"type":"attachment","uuid":"native-array","timestamp":"2026-08-05T06:00:09Z","attachment":{"type":"queued_command","prompt":[{"type":"text","text":"array message text"}],"commandMode":"prompt"}}`, + `{"type":"queue-operation","operation":"enqueue","timestamp":"2026-08-05T06:00:10Z","sessionId":"S","content":"after the array"}`, + `{"type":"queue-operation","operation":"dequeue","timestamp":"2026-08-05T06:00:11Z","sessionId":"S"}`, + `{"type":"user","uuid":"echo-user-2","message":{"role":"user","content":"after the array"}}`, + ].join("\n") ~ "\n"; + write(jsonlPath, jsonl); + + td.history.reset(Watermark.atBytes(getSize(jsonlPath))); + + HistoryEventPipelineHost host; + host.getTask = (int t) => t == tid ? &td : null; + host.resolveTaskHistory = (int t) => TaskHistoryResolution.access( + HistoryAccess(agent, profile, td.agentSessionId, projectPath, jsonlPath)); + host.injectAgentNameIntoSessionInit = (string translated, string agentName) => translated; + host.normalizeKnownSystemMessageMeta = (string translated, int t) => translated; + host.makeTaskDiagnosticEventJson = (string subject, string body) => ""; + host.sendToSubscribed = (int t, Data d) {}; + host.subscribe = (WebSocketAdapter ws, int t) {}; + host.sendHistoryOperations = (WebSocketAdapter ws, int t) {}; + host.broadcastHistoryOperations = (int t) {}; + host.sendReplaySupplementalState = (WebSocketAdapter ws, int t) {}; + host.onHistorySubscribed = (int t) {}; + host.updateClaudeUsageFromEvent = (int t, string translated) => false; + host.planBroadcast = (int t, TranslatedEvent ev) => HistoryBroadcastPlan.init; + host.noteLiveBoundaryCandidate = (int t, size_t seq, string ev, string raw, + int sourceLine, bool isContextBootstrap) {}; + host.configuredAgentNames = () => cast(string[]) null; + + auto pipeline = new HistoryEventPipeline(host); + pipeline.ensureHistoryLoaded(tid); + assert(td.history.isLoaded); + + string[] events; + foreach (i, ref ev; td.history) + events ~= cast(string) ev.toGC(); + + size_t deliveredConfirmations = 0, removedConfirmations = 0, canonicalMessages = 0; + foreach (ev; events) + { + if (ev.canFind(`"user_message/consumed"`)) + { + if (ev.canFind(`"consumed_as":"steering"`)) + deliveredConfirmations++; + if (ev.canFind(`"consumed_as":"removed"`)) + removedConfirmations++; + } + if (ev.canFind(`"cc-queued-command"`)) + canonicalMessages++; + } + + // both absorbed messages are confirmed as consumed, each pointing at the + // canonical message that carries the delivery + assert(deliveredConfirmations == 3, "absorbed messages must be confirmed"); + assert(canonicalMessages == 3, "each absorbed delivery becomes one canonical message"); + // only the queue-cleared one keeps the not-delivered outcome + assert(removedConfirmations == 1, "an undelivered remove stays undelivered"); + // the dequeue-flow attachment renders nothing: the echo is the message + size_t echoedMessages = 0; + foreach (ev; events) + if (ev.canFind(`"echoed"`) && ev.canFind(`"item/started"`)) + echoedMessages++; + foreach (ev; events) + assert(!ev.canFind("native-echoed"), + "a dequeue-flow attachment must not surface"); + assert(echoedMessages == 2, + "the echoed message appears as its provisional and its echo only"); + // the array message is delivered via empty-to-empty pairing + bool sawArrayDelivery, sawAfterArrayPairing; + foreach (ev; events) + { + if (ev.canFind(`"consumed_as":"steering"`) && ev.canFind("native-array")) + sawArrayDelivery = true; + // the turn opener after the array message pairs with its own entry, + // not a stranded one + if (ev.canFind(`"user_message/consumed"`) && ev.canFind(`"turn_start"`) + && ev.canFind(`"echo-user-2"`)) + sawAfterArrayPairing = true; + } + assert(sawArrayDelivery, + "an array-valued absorbed message must still deliver"); + assert(sawAfterArrayPairing, + "the pairing after an array message must not shift"); + + foreach (ev; events) + if (ev.canFind(`"consumed_as":"steering"`) && ev.canFind(`"first"`)) + assert(ev.canFind(`"native_uuid":"native-first"`)); + // no delivered message may be marked undelivered + foreach (ev; events) + if (ev.canFind(`"consumed_as":"removed"`)) + assert(!ev.canFind("native-"), "a settled removal names no delivery"); +} + unittest { import cydo.agent.contract : PersistedHistoryBoundaryKind; diff --git a/web/src/sessionReducer.test.ts b/web/src/sessionReducer.test.ts index c4cfaf61..b241ca50 100644 --- a/web/src/sessionReducer.test.ts +++ b/web/src/sessionReducer.test.ts @@ -1264,3 +1264,77 @@ describe("cydo/task_spawned reducer", () => { expect(s.pendingCydoTaskItemIds).toEqual([]); }); }); + +describe("mid-turn absorption", () => { + const consumed = (event: { + uuid: string; + consumed_as: string; + native_uuid?: string; + }) => asEvent({ type: "user_message/consumed", ...event }); + + it("hands a queued provisional over to the mid-turn delivery", () => { + // the enqueue-emitted provisional sits in the agent's queue; the mid-turn + // delivery confirms it and hands over to the canonical message + let state = reduceMessage( + makeState(), + asEvent({ + type: "item/started", + item_id: "synthetic-user", + item_type: "user_message", + uuid: "enqueue-3", + pending: true, + content: [{ type: "text", text: "sent while busy" }], + }), + ); + expect(state.messages).toHaveLength(1); + expect(state.messages[0]?.removed).toBeUndefined(); + + state = reduceMessage( + state, + consumed({ + uuid: "enqueue-3", + native_uuid: "native-9", + consumed_as: "steering", + }), + ); + // the canonical message follows, so the provisional is gone entirely + expect(state.messages.filter((m) => m.type === "user")).toHaveLength(0); + + state = reduceMessage( + state, + asEvent({ + type: "item/started", + item_id: "cc-queued-command", + item_type: "user_message", + uuid: "native-9", + content: [{ type: "text", text: "sent while busy" }], + }), + ); + const users = state.messages.filter((m) => m.type === "user"); + expect(users).toHaveLength(1); + expect(users[0]?.uuid).toBe("native-9"); + expect(users[0]?.removed).toBeUndefined(); + }); + + it("keeps the not-delivered badge for a queue clear", () => { + let state = reduceMessage( + makeState(), + asEvent({ + type: "item/started", + item_id: "synthetic-user", + item_type: "user_message", + uuid: "enqueue-5", + pending: true, + content: [{ type: "text", text: "never sent" }], + }), + ); + // no delivery record: the confirmation carries no native uuid + state = reduceMessage( + state, + consumed({ uuid: "enqueue-5", consumed_as: "removed" }), + ); + const users = state.messages.filter((m) => m.type === "user"); + expect(users).toHaveLength(1); + expect(users[0]?.removed).toBe(true); + }); +});