feat(sessions): nudge alive tasks whose transcript ended mid-turn

source/cydo/domain/storage/persistence.d

@@ -134,6 +134,12 @@ struct Persistence
 			"    has_messages INTEGER NOT NULL DEFAULT 1," ~
 			"    PRIMARY KEY (driver, profile_root, session_id)" ~
 			");",
+			// Migration 23: whether the transcript ends inside an open turn.
+			// Maintained on every translated event (set on turn activity,
+			// cleared on turn/result and process/exit), so a restart or crash
+			// landing mid-turn is visible at resume time regardless of what
+			// the status field recorded; such tasks get the restart nudge.
+			"ALTER TABLE tasks ADD COLUMN turn_open INTEGER NOT NULL DEFAULT 0;",
 		]);
 
 		// In CI, disable durability to speed up tests. This trades crash-safety
@@ -219,6 +225,11 @@ struct Persistence
 		db.stmt!"UPDATE tasks SET status = ? WHERE tid = ?".exec(status, tid);
 	}
 
+	void setTurnOpen(int tid, bool open)
+	{
+		db.stmt!"UPDATE tasks SET turn_open = ? WHERE tid = ?".exec(open ? 1 : 0, tid);
+	}
+
 	void promoteImportableTask(int tid, string workspace)
 	{
 		db.stmt!"UPDATE tasks SET workspace = ?, status = 'completed' WHERE tid = ? AND status = 'importable'"
@@ -292,6 +303,7 @@ struct Persistence
 		long lastActive;
 		string entryPoint;
 		bool needsAttention;
+		bool turnOpen;
 	}
 
 	TaskRow[] loadTasks()
@@ -300,11 +312,11 @@ struct Persistence
 		foreach (int tid, string agentSessionId, string description, string taskType,
 			int parentTid, string relationType, string workspace, string projectPath,
 			int worktreeTid, string taskStartHead, string title, string status, string agentName, int archived, string draft,
-			string resultText, long createdAt, long lastActive, string entryPoint, int needsAttention;
-			db.stmt!"SELECT tid, COALESCE(agent_session_id,''), COALESCE(description,''), COALESCE(task_type,'blank'), COALESCE(parent_tid,0), COALESCE(relation_type,''), COALESCE(workspace,''), COALESCE(project_path,''), COALESCE(worktree_tid,0), COALESCE(task_start_head,''), COALESCE(title,''), COALESCE(status,'completed'), COALESCE(agent_type,'claude'), COALESCE(archived,0), COALESCE(draft,''), COALESCE(result_text,''), COALESCE(created_at,0), COALESCE(last_active,0), COALESCE(entry_point,''), COALESCE(needs_attention,0) FROM tasks".iterate())
+			string resultText, long createdAt, long lastActive, string entryPoint, int needsAttention, int turnOpen;
+			db.stmt!"SELECT tid, COALESCE(agent_session_id,''), COALESCE(description,''), COALESCE(task_type,'blank'), COALESCE(parent_tid,0), COALESCE(relation_type,''), COALESCE(workspace,''), COALESCE(project_path,''), COALESCE(worktree_tid,0), COALESCE(task_start_head,''), COALESCE(title,''), COALESCE(status,'completed'), COALESCE(agent_type,'claude'), COALESCE(archived,0), COALESCE(draft,''), COALESCE(result_text,''), COALESCE(created_at,0), COALESCE(last_active,0), COALESCE(entry_point,''), COALESCE(needs_attention,0), COALESCE(turn_open,0) FROM tasks".iterate())
 		{
 			// tasks.agent_type stores the configured agent name from config.agents.
-			result ~= TaskRow(tid, agentSessionId, description, taskType, parentTid, relationType, workspace, projectPath, worktreeTid, taskStartHead, title, status, agentName, archived != 0, draft, resultText, createdAt, lastActive, entryPoint, needsAttention != 0);
+			result ~= TaskRow(tid, agentSessionId, description, taskType, parentTid, relationType, workspace, projectPath, worktreeTid, taskStartHead, title, status, agentName, archived != 0, draft, resultText, createdAt, lastActive, entryPoint, needsAttention != 0, turnOpen != 0);
 		}
 		return result;
 	}
@@ -610,7 +622,7 @@ unittest
 	int userVersion;
 	foreach (int value; persistence.db.stmt!"PRAGMA user_version".iterate())
 		userVersion = value;
-	assert(userVersion == 22);
+	assert(userVersion == 23);
 
 	auto rows = persistence.loadTasks();
 	assert(rows.length == 1);
@@ -630,6 +642,7 @@ unittest
 		"worktree_path", "has_worktree", "agent_type", "archived", "draft",
 		"result_text", "created_at", "last_active", "worktree_tid", "entry_point",
 		"needs_attention", "task_start_head",
+		"turn_open",
 	]);
 
 	persistence.upsertSessionMetaCache("claude", "/profiles/one", "same-id", 1,

source/cydo/domain/tasks/model.d

@@ -557,6 +557,9 @@ struct TaskData
 	bool archived;
 	long createdAt;    // StdTime; 0 = not set
 	long lastActive;   // StdTime; 0 = not set
+	/// the transcript ends inside an open turn; persisted, so a restart or
+	/// crash landing mid-turn is visible to resume regardless of status
+	bool turnOpen;
 
 	/// Git repository root for the selected project.
 	/// Falls back to projectPath if git resolution fails.

source/cydo/server/app.d

@@ -544,6 +544,9 @@ class App
 			onHistorySubscribed: &onHistorySubscribed,
 			updateClaudeUsageFromEvent: &updateClaudeUsageFromEvent,
 			planBroadcast: &planHistoryBroadcast,
+			persistTurnOpen: (int tid, bool open) {
+				persistence.setTurnOpen(tid, open);
+			},
 		));
 		derivedTextJobs = new DerivedTextJobs(DerivedTextJobsHost(
 			getTask: (int tid) => tid in tasks ? &tasks[tid] : null,
@@ -838,6 +841,7 @@ class App
 			td.createdAt = row.createdAt;
 			td.lastActive = row.lastActive;
 			td.needsAttention = row.needsAttention;
+			td.turnOpen = row.turnOpen;
 			td.titleGenDone = row.title.length > 0;
 			auto rowTid = row.tid;
 			tasks[rowTid] = move(td);

source/cydo/workflow/history/pipeline.d

@@ -58,6 +58,7 @@ struct HistoryEventPipelineHost
 	void delegate(int tid) onHistorySubscribed;
 	bool delegate(int tid, string translated) updateClaudeUsageFromEvent;
 	HistoryBroadcastPlan delegate(int tid, TranslatedEvent ev) planBroadcast;
+	void delegate(int tid, bool open) persistTurnOpen;
 }
 
 class HistoryEventPipeline
@@ -505,6 +506,24 @@ class HistoryEventPipeline
 		td = host_.getTask(tid);
 		if (td is null)
 			return 0;
+		// witness the turn lifecycle for restart recovery: a turn-scoped
+		// event (item/*, turn/*) marks the transcript mid-turn until a
+		// turn/result or process exit closes it, and session bootstrap or
+		// other non-turn events leave it unchanged, so a restart's own
+		// resume traffic cannot re-open an idle task. persisted on change,
+		// so a kill at any moment leaves the truth on disk
+		auto closesTurn = isTurnResultEvent(ev.translated)
+			|| isProcessExitEvent(ev.translated);
+		auto opensTurn = !closesTurn && (ev.translated.canFind(`"type":"item/`)
+			|| ev.translated.canFind(`"type":"item\/`)
+			|| ev.translated.canFind(`"type":"turn/`)
+			|| ev.translated.canFind(`"type":"turn\/`));
+		if ((closesTurn || opensTurn) && td.turnOpen != opensTurn)
+		{
+			td.turnOpen = opensTurn;
+			if (host_.persistTurnOpen !is null)
+				host_.persistTurnOpen(tid, opensTurn);
+		}
 		if (merged)
 			seq = td.history.isLoaded ? td.history.length - 1 : cast(size_t) -1;
 		else
@@ -1541,3 +1560,72 @@ unittest
 	assert(sawSteering, "steering confirmation missing");
 	assert(sawRemoved, "removed confirmation missing");
 }
+
+unittest
+{
+	// the turn witness: a turn-scoped event (item/*, turn/*) opens the turn,
+	// a turn result or process exit closes it, session events change nothing,
+	// and only changes hit disk, so a kill at any moment leaves the truth
+	// persisted for resume
+	import cydo.domain.tasks.model : Watermark;
+
+	enum tid = 7;
+	auto td = TaskData(tid, "local", "/tmp/cydo-turn-witness-project");
+	td.history.reset(Watermark.init);
+
+	bool[] persisted;
+	HistoryEventPipelineHost host;
+	host.getTask = (int t) => t == tid ? &td : null;
+	host.normalizeKnownSystemMessageMeta = (string translated, int t) => translated;
+	host.sendToSubscribed = (int t, Data data) {};
+	host.updateClaudeUsageFromEvent = (int t, string translated) => false;
+	host.planBroadcast = (int t, TranslatedEvent ev) {
+		HistoryBroadcastPlan plan;
+		plan.currentEvent = ev;
+		return plan;
+	};
+	host.persistTurnOpen = (int t, bool open) { persisted ~= open; };
+	auto pipeline = new HistoryEventPipeline(host);
+
+	TranslatedEvent make(string translated)
+	{
+		TranslatedEvent ev;
+		ev.translated = translated;
+		return ev;
+	}
+
+	assert(!td.turnOpen, "a fresh task starts with no open turn");
+
+	// activity opens the turn and persists the change once
+	pipeline.broadcastTask(tid, make(`{"type":"item/started","item_type":"text","item_id":"a1","text":"hi"}`));
+	assert(td.turnOpen, "turn activity must open the turn");
+	assert(persisted == [true], "the open must be persisted exactly once");
+
+	// further activity changes nothing on disk
+	pipeline.broadcastTask(tid, make(`{"type":"item/started","item_type":"user_message","item_id":"u1","text":"echo"}`));
+	assert(persisted == [true], "an already-open turn must not rewrite the flag");
+
+	// the turn result closes it
+	pipeline.broadcastTask(tid, make(`{"type":"turn/result","cost_usd":0}`));
+	assert(!td.turnOpen, "a turn result must close the turn");
+	assert(persisted == [true, false]);
+
+	// a process exit while already closed writes nothing further
+	pipeline.broadcastTask(tid, make(`{"type":"process/exit","code":0}`));
+	assert(persisted == [true, false], "closing an already-closed turn must not write");
+
+	// session bootstrap events are not turn activity in either direction:
+	// a restart's resume traffic must not re-open an idle task
+	pipeline.broadcastTask(tid, make(`{"type":"session/metadata","model":"m"}`));
+	assert(!td.turnOpen && persisted == [true, false],
+		"a session event must not open a closed turn");
+
+	// an exit that lands mid-turn also closes it, and a session event in
+	// between changes nothing
+	pipeline.broadcastTask(tid, make(`{"type":"item/started","item_type":"text","item_id":"a2","text":"more"}`));
+	pipeline.broadcastTask(tid, make(`{"type":"session/metadata","model":"m"}`));
+	assert(td.turnOpen, "a session event must not close an open turn");
+	pipeline.broadcastTask(tid, make(`{"type":"process/exit","code":1}`));
+	assert(!td.turnOpen, "a process exit must close an open turn");
+	assert(persisted == [true, false, true, false]);
+}

source/cydo/workflow/sessions/task_runner.d

@@ -1307,7 +1307,18 @@ public:
 			}
 			else if (status == "alive")
 			{
-				resumeTask(tid).ignoreResult();
+				if (td.turnOpen)
+				{
+					// killed mid-turn: the status snapshot can miss this (a
+					// turn started by the agent itself, e.g. a background-task
+					// notification, never passes through a send that marks the
+					// task active), but the persisted turn witness cannot
+					infof("resumeInFlightTasks: tid=%d transcript ends mid-turn, resuming with restart nudge",
+						tid);
+					resumeActiveTask(tid);
+				}
+				else
+					resumeTask(tid).ignoreResult();
 			}
 		}
 	}

source/cydo/workflow/tasks/subtask_delivery.d

@@ -195,7 +195,10 @@ public:
 
 			enum nudgeBody = "Your session was interrupted by a harness restart. "
 				~ "Continue from where you left off. If you had a tool call in progress "
-				~ "(Task, Handoff, SwitchMode, or any other tool), retry it.";
+				~ "(Task, Handoff, SwitchMode, or any other tool), retry it. "
+				~ "Background work (shells, subagents, watchers) did not survive the "
+				~ "restart: check on anything you had running and relaunch what is "
+				~ "still needed before ending your turn.";
 			try
 			{
 				host_.sendKnownSystemMessage(tid, KnownSystemMessageKind.restartNudge,