From 608a864fe561174f85a227347bfd31fd24952c0e Mon Sep 17 00:00:00 2001 From: Antisophy <293439221+Antisophy@users.noreply.github.com> Date: Fri, 28 Aug 2026 16:57:18 -0700 Subject: [PATCH 1/2] feat(sessions): nudge alive tasks whose transcript ended mid-turn A restart or crash that lands mid-turn is only recovered when the task was recorded "active": that resume path relaunches the process and sends the restart nudge telling the agent to retry its interrupted tool call. A turn started by the agent itself (a background-task notification inside the CLI) never passes through a send that marks the task active, so the shutdown snapshot says "alive", the resume relaunches silently, and the agent sits idle on its interrupted work until a human notices. Witness the turn lifecycle directly instead of trusting the status: a turn-scoped event (item/*, turn/*) marks the task's transcript mid-turn until a turn/result or process exit closes it, while session bootstrap and other non-turn events leave the witness unchanged, so a restart's own resume traffic cannot re-open an idle task. The flag is persisted on change, so a kill at any moment leaves the truth on disk. The alive resume branch sends the restart nudge when the witness says mid-turn and stays silent otherwise, so idle tasks cost nothing extra. --- source/cydo/domain/storage/persistence.d | 21 ++++- source/cydo/domain/tasks/model.d | 3 + source/cydo/server/app.d | 4 + source/cydo/workflow/history/pipeline.d | 88 +++++++++++++++++++++ source/cydo/workflow/sessions/task_runner.d | 13 ++- 5 files changed, 124 insertions(+), 5 deletions(-) diff --git a/source/cydo/domain/storage/persistence.d b/source/cydo/domain/storage/persistence.d index cc90a011..3106da5a 100644 --- a/source/cydo/domain/storage/persistence.d +++ b/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, diff --git a/source/cydo/domain/tasks/model.d b/source/cydo/domain/tasks/model.d index 4caacf27..6f629bca 100644 --- a/source/cydo/domain/tasks/model.d +++ b/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. diff --git a/source/cydo/server/app.d b/source/cydo/server/app.d index c127f5ba..a5ad2b49 100644 --- a/source/cydo/server/app.d +++ b/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); diff --git a/source/cydo/workflow/history/pipeline.d b/source/cydo/workflow/history/pipeline.d index 35a77206..506b8a4a 100644 --- a/source/cydo/workflow/history/pipeline.d +++ b/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]); +} diff --git a/source/cydo/workflow/sessions/task_runner.d b/source/cydo/workflow/sessions/task_runner.d index 2d525730..07b29f57 100644 --- a/source/cydo/workflow/sessions/task_runner.d +++ b/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(); } } } From 9a8a97517f09a3a5a373ce07f0d80717efa14f87 Mon Sep 17 00:00:00 2001 From: Antisophy <293439221+Antisophy@users.noreply.github.com> Date: Fri, 28 Aug 2026 17:03:03 -0700 Subject: [PATCH 2/2] feat(sessions): tell nudged agents to check on killed background work The restart nudge covered an interrupted tool call but said nothing about background work, and an agent whose pending work was a background shell or watcher rather than an in-flight tool call would conclude nothing was pending and end its turn; the background jobs died with the restart and their completion notifications never fire, so the task orphans. Name that case explicitly: check what was running, relaunch what is still needed, and do not end the turn without doing so. --- source/cydo/workflow/tasks/subtask_delivery.d | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/source/cydo/workflow/tasks/subtask_delivery.d b/source/cydo/workflow/tasks/subtask_delivery.d index 6e2641d4..4d4d4e7e 100644 --- a/source/cydo/workflow/tasks/subtask_delivery.d +++ b/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,