fix(history): treat mid-turn absorption as delivery, not withdrawal

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)

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

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,