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