refactor(interactive): remove streaming interactive stack (superseded by ConPTY)
The embedded ConPTY terminal replaced the in-app streaming interactive session, so delete the dead stack: StreamingClaudeSession, InteractiveSessionService, ProcessClaudeStreamTransport, IClaudeStreamTransport, ILiveSession, LiveSessionRegistry, IdleSessionReaper (+ WorkerConfig.InteractiveIdleTimeoutMinutes), the WorkerHub interactive methods + HubBroadcaster events, IWorkerClient interactive members, the TaskMonitorViewModel composer + SessionTerminalView composer markup, and the old 'Run interactively' entry. AskUser/PendingQuestionRegistry, the autonomous path, planning, ResumeTaskInTerminal, and all ConPTY code are kept. Localization pruned.
This commit is contained in:
mika kuns
+1
-7
@@ -17,13 +17,7 @@ Kein Code-Aufwand, nur Durchspielen mit explizit notiertem Pass-Kriterium. Der G
|
||||
- **UI-Sichtprüfung (neu, 2026-06-10, nach Refactoring-Merges):** Detail-Insel komplett durchklicken (Output/Git/Session-Tabs, Merge-Sektion, Agent-Settings-Overrides, Prep-Panel) — `DetailsIslandViewModel` wurde in Sektions-VMs aufgeteilt, Bindings angepasst. Außerdem: DiffModal-Fehler-State „Diff nicht mehr verfügbar" (Commit-Range ohne aufgezeichnete Commits) und der In-App-Konflikt-Resolver (Hub-Methoden umbenannt).
|
||||
- **UI-Sichtprüfung (neu, 2026-06-19, Rider-Style 3-Pane Merge-Editor):** Echten Konflikt auslösen (Single-Task-Approve mit Konflikt **und** Planning-Unit-Merge) und prüfen: drei Panes (Ours read-only | Result editierbar | Theirs read-only), Konfliktblöcke rot / aufgelöst grün in allen Panes, Inline-Accept `›`/`‹` in den Zwischen-Guttern landen die jeweilige Seite im Result, nur Konfliktregionen im Result editierbar (Stable read-only), synchrones vertikales Scrollen, File-Switcher bei mehreren Dateien, `M conflicts · K resolved`-Readout, Continue erst bei allen Konflikten gelöst, Binär-Guard. **Bekannte Kanten:** (1) Konflikt mit leerer Ours-Seite → Result-Region ist null-lang (Gutter via 1-Zeichen-Probe positioniert, Accept funktioniert; nur Hand-Tippen in die leere Region ist fummelig). (2) Gutter-Y nutzt `TranslatePoint` vom Result-`TextView` — bei sehr hohen Fenstern / großen Scrollständen die Ausrichtung gegenprüfen. (3) Blöcke richten sich nur über Stable-Text aus; nach einem Konflikt mit unterschiedlicher Zeilenzahl je Seite driften nachfolgende Blöcke vertikal (aligned/virtual-space Scroll ist bewusst zurückgestellt).
|
||||
- **Worker-Autostart am Gerät:** Logoff/Logon-Autostart, Update-Pfad, Uninstall entfernt die Startup-`.lnk`.
|
||||
- **In-App Interactive Sessions (neu, 2026-06-26):** ersetzt den externen `wt`-„Run interactively"-Launch durch einen In-App-Streaming-Chat (`StreamingClaudeSession`, `claude --input-format stream-json`). Real-CLI-Smoke (kein xUnit, kein Claude in Tests):
|
||||
- Task rechtsklick → „Run interactively" startet **keinen** Terminal mehr; der Stream erscheint im Detail-Output-Tab des (selektierten) Tasks und als Monitor in Mission Control.
|
||||
- Composer: Nachricht tippen + Enter/Send → erscheint sofort als `log-user`-Zeile in **Akzentfarbe** (via `LogKindForegroundConverter`, lokale Bindung schlägt den dim Style), Claude antwortet im selben Prozess.
|
||||
- **Senden während Claude arbeitet = Queue (Default):** die Nachricht wird gepuffert und beim `result` des laufenden Turns abgeschickt (kein Interrupt). Mehrere Queue-Nachrichten FIFO, eine pro Turn. Gequeute Nachrichten erscheinen in einem **Pending-Streifen über der Eingabezeile** (⧗-Liste, via `InteractiveQueueChanged`); eine Nachricht landet erst im Transkript (`log-user`-Zeile via `InteractiveMessageSent`), wenn sie tatsächlich an Claude zugestellt wird. Der seeded Erst-Prompt erscheint als erste User-Zeile. Jede gequeute Zeile hat ein **✕ zum Entfernen** (`RemoveQueuedInteractiveMessage`, by-text first-match; Worker re-broadcastet die Queue).
|
||||
- **Interrupt opt-in:** der kleine ■-Stop-Button neben Send unterbricht den laufenden Turn (`control_request`/`interrupt`, verifiziert mit CLI 2.1.191; Abbruch-`result` = `error_during_execution`, als Turn-Ende behandelt) — danach flusht die ggf. gequeute Nachricht im selben Prozess mit erhaltenem Kontext. Stop-Button ist immer sichtbar solange live (Interrupt im Idle ist ein No-op; Turn-in-flight wird nicht in die UI gebroadcastet).
|
||||
- Session-Ende: Prozess-Exit/Stop → `InteractiveSessionEnded`, Composer verschwindet, Monitor wird „done".
|
||||
- **Sicht-Konsistenz:** Mission-Control-Composer (SessionTerminalView-Bottom-Row mit Send-Button) vs. Detail-Composer (WorkConsole-Shell-Prompt `❯ … [Send]`) sehen unterschiedlich aus — ggf. angleichen.
|
||||
- **In-App Interactive Sessions (2026-06-26, REMOVED 2026-07-23):** der In-App-Streaming-Chat (`StreamingClaudeSession`, Composer/Queue auf `TaskMonitorViewModel`/`SessionTerminalView`) wurde komplett entfernt und durch die **embedded ConPTY**-Sessions ersetzt (echte `claude`-TUI im UI-Prozess, siehe `docs/superpowers/specs/2026-07-23-conpty-interactive-sessions-design.md`). Kein offener Punkt mehr — nur zur Historie.
|
||||
- **Pick up in terminal (neu, 2026-07-01):** neue Aktion, die die Claude-Session einer Task per `claude --resume <id>` in einem **echten** `wt`-Terminal fortsetzt (echte TUI: Permission-Prompts/Fragen inklusive) — bewusst NICHT der In-App-Streaming-Chat. Real-CLI-Smoke (kein Claude in Tests):
|
||||
- Kontextmenü einer Task in **WaitingForReview** oder **Failed** → „Pick up in terminal" sowie der Terminal-Button (Icon `ArrowOut`) im Detail-Header sind sichtbar; bei anderen Status (Idle/Running/Queued/Done) NICHT.
|
||||
- Klick → neues Windows-Terminal im Worktree-Verzeichnis der Task, Claude nimmt die letzte Session mit erhaltenem Kontext wieder auf.
|
||||
|
||||
@@ -128,7 +128,6 @@
|
||||
"ctxMarkAs": "Markieren als",
|
||||
"ctxMarkDone": "Erledigt",
|
||||
"ctxMarkCancelled": "Abgebrochen",
|
||||
"ctxRunInteractively": "Interaktiv ausführen",
|
||||
"ctxOpenConPtySession": "ConPTY-Sitzung öffnen",
|
||||
"ctxPickUpInTerminal": "Im Terminal fortsetzen",
|
||||
"ctxOpenPlanningSession": "Planungssitzung öffnen",
|
||||
@@ -247,15 +246,7 @@
|
||||
"chipDone": "FERTIG",
|
||||
"chipFailed": "FEHLGESCHLAGEN",
|
||||
"reviewContinueTip": "Dieses Feedback senden und die Aufgabe erneut ausführen",
|
||||
"reviewResetTip": "Alle Änderungen verwerfen und die Aufgabe auf Leerlauf zurücksetzen",
|
||||
"composer": {
|
||||
"placeholder": "Nachricht an die Sitzung…",
|
||||
"send": "Senden",
|
||||
"stop": "Sitzung beenden",
|
||||
"interrupt": "Aktuellen Zug unterbrechen",
|
||||
"queued": "Wartet — wird nach dem aktuellen Zug gesendet",
|
||||
"unqueue": "Aus Warteschlange entfernen"
|
||||
}
|
||||
"reviewResetTip": "Alle Änderungen verwerfen und die Aufgabe auf Leerlauf zurücksetzen"
|
||||
},
|
||||
"missionControl": {
|
||||
"openInApp": "In App öffnen",
|
||||
@@ -513,7 +504,7 @@
|
||||
"taskStatus": { "idle": "Leerlauf", "queued": "In Warteschlange", "running": "Läuft", "waitingForReview": "Wartet auf Prüfung", "waitingForChildren": "Wartet auf Verbesserungen", "done": "Fertig", "failed": "Fehlgeschlagen", "cancelled": "Abgebrochen", "parked": "Geparkt" },
|
||||
"planningBadge": { "active": "PLANUNG", "finalized": "GEPLANT" },
|
||||
"taskRow": { "createdPrefix": "Erstellt {0}", "stepsText": "{0}/{1} Schritte" },
|
||||
"tasksIsland": { "completedHeader": "ABGESCHLOSSEN", "completedHeaderCount": "ABGESCHLOSSEN · {0}", "runInteractiveFailed": "Interaktiv ausführen fehlgeschlagen: {0}", "planningOpenFailed": "Planungssitzung konnte nicht geöffnet werden: {0}", "pickUpInTerminalFailed": "Im Terminal fortsetzen fehlgeschlagen: {0}", "approveFailed": "Genehmigen & Mergen fehlgeschlagen: {0}" },
|
||||
"tasksIsland": { "completedHeader": "ABGESCHLOSSEN", "completedHeaderCount": "ABGESCHLOSSEN · {0}", "planningOpenFailed": "Planungssitzung konnte nicht geöffnet werden: {0}", "pickUpInTerminalFailed": "Im Terminal fortsetzen fehlgeschlagen: {0}", "approveFailed": "Genehmigen & Mergen fehlgeschlagen: {0}" },
|
||||
"diff": { "loadFailed": "Diff konnte nicht geladen werden: {0}", "noChanges": "Keine Änderungen anzuzeigen.", "unavailable": "Diff nicht mehr verfügbar — Commit-Bereich unvollständig." },
|
||||
"planningDiff": { "hubError": "Kombinierte Vorschau konnte nicht erstellt werden (Hub-Fehler).", "conflict": "Kombinierte Vorschau nicht möglich: Teilaufgabe {0} steht im Konflikt mit einer früheren Teilaufgabe ({1} Dateien)." },
|
||||
"merge": { "commitMessage": "Merge-Aufgabe: {0}", "workerOfflineBranches": "Worker offline — Branches können nicht aufgelistet werden.", "loadBranchesFailed": "Branches konnten nicht geladen werden: {0}", "merged": "Zusammengeführt.", "conflict": "Merge-Konflikt — Ziel-Branch wiederhergestellt. Manuell oder über Fortsetzen lösen, dann erneut versuchen.", "blocked": "Blockiert: {0}", "unknownStatus": "Unbekannter Status: {0}", "mergeFailed": "Merge fehlgeschlagen: {0}" },
|
||||
|
||||
@@ -128,7 +128,6 @@
|
||||
"ctxMarkAs": "Mark as",
|
||||
"ctxMarkDone": "Done",
|
||||
"ctxMarkCancelled": "Cancelled",
|
||||
"ctxRunInteractively": "Run interactively",
|
||||
"ctxOpenConPtySession": "Open ConPTY session",
|
||||
"ctxPickUpInTerminal": "Pick up in terminal",
|
||||
"ctxOpenPlanningSession": "Open planning Session",
|
||||
@@ -247,15 +246,7 @@
|
||||
"chipDone": "DONE",
|
||||
"chipFailed": "FAILED",
|
||||
"reviewContinueTip": "Send this feedback and re-run the task",
|
||||
"reviewResetTip": "Discard all changes and reset the task to Idle",
|
||||
"composer": {
|
||||
"placeholder": "Message the session…",
|
||||
"send": "Send",
|
||||
"stop": "Stop session",
|
||||
"interrupt": "Interrupt current turn",
|
||||
"queued": "Queued — sends after the current turn",
|
||||
"unqueue": "Remove from queue"
|
||||
}
|
||||
"reviewResetTip": "Discard all changes and reset the task to Idle"
|
||||
},
|
||||
"missionControl": {
|
||||
"openInApp": "Open in app",
|
||||
@@ -513,7 +504,7 @@
|
||||
"taskStatus": { "idle": "Idle", "queued": "Queued", "running": "Running", "waitingForReview": "Waiting for Review", "waitingForChildren": "Waiting for Improvements", "done": "Done", "failed": "Failed", "cancelled": "Cancelled", "parked": "Parked" },
|
||||
"planningBadge": { "active": "PLANNING", "finalized": "PLANNED" },
|
||||
"taskRow": { "createdPrefix": "Created {0}", "stepsText": "{0}/{1} steps" },
|
||||
"tasksIsland": { "completedHeader": "COMPLETED", "completedHeaderCount": "COMPLETED · {0}", "runInteractiveFailed": "Run interactively failed: {0}", "planningOpenFailed": "Couldn't open planning session: {0}", "pickUpInTerminalFailed": "Pick up in terminal failed: {0}", "approveFailed": "Approve & merge failed: {0}" },
|
||||
"tasksIsland": { "completedHeader": "COMPLETED", "completedHeaderCount": "COMPLETED · {0}", "planningOpenFailed": "Couldn't open planning session: {0}", "pickUpInTerminalFailed": "Pick up in terminal failed: {0}", "approveFailed": "Approve & merge failed: {0}" },
|
||||
"diff": { "loadFailed": "Failed to load diff: {0}", "noChanges": "No changes to show.", "unavailable": "Diff no longer available — commit range incomplete." },
|
||||
"planningDiff": { "hubError": "Could not build combined preview (hub error).", "conflict": "Cannot build combined preview: subtask {0} conflicts with an earlier subtask ({1} files)." },
|
||||
"merge": { "commitMessage": "Merge task: {0}", "workerOfflineBranches": "Worker offline — cannot list branches.", "loadBranchesFailed": "Failed to load branches: {0}", "merged": "Merged.", "conflict": "Merge conflict — target branch restored. Resolve manually or via Continue, then retry.", "blocked": "Blocked: {0}", "unknownStatus": "Unknown status: {0}", "mergeFailed": "Merge failed: {0}" },
|
||||
|
||||
@@ -35,7 +35,7 @@ Design/ — Tokens.axaml (design tokens; merged before styles) + IslandStyle
|
||||
|
||||
- **IslandsShellViewModel** — root coordinator; owns the three island VMs and the `WorkerClient`, wires cross-island events (selection, notes/prep mode, conflict resolution), owns connection state, the update banner, the inline worker-log strip (clickable → Log Visualizer overlay via `OpenLogVisualizerCommand`; `FlashFooterError` surfaces UI-action failures + the worker's Serilog Warn/Error there), responsive-layout flags (`ShowLists`/`ShowDetails` by window width), `PrimeStatus` flash, and the modal openers (About, RepoImport, WeeklyReport, WorktreesOverview, WorkerConnection help, LogVisualizer) plus `RestartWorkerAsync`/`CheckForUpdatesAsync`. Hosts `UpdateCheckService`.
|
||||
- **ListsIslandViewModel** — smart lists (My Day, Important, Planned, virtual queued/running/review), user lists, selection, list CRUD, drag-reorder, badge counts, opens list settings / repo import / worktrees overview, `OpenInExplorer`/`OpenInTerminal`.
|
||||
- **TasksIslandViewModel** — open/overdue/completed groups for the selected list with hierarchy-aware regrouping; task CRUD, drag-reorder, toggle done/star, schedule, enqueue/dequeue, cancel; review actions (approve, reject-rerun, reject-park, cancel); planning session lifecycle (open/resume/discard/finalize, `QueuePlanningSubtasksAsync`); `RunInteractivelyAsync`, `RefineTask`; MyDay extras (`IsMyDayList`, `ClearDayCommand`, `ShowPrepLogCommand`) and the pinned Notes pseudo-row (`ShowNotesRow`, `OpenNotesCommand`). Raises `NotesRequested`/`PrepRequested` events consumed by the shell.
|
||||
- **TasksIslandViewModel** — open/overdue/completed groups for the selected list with hierarchy-aware regrouping; task CRUD, drag-reorder, toggle done/star, schedule, enqueue/dequeue, cancel; review actions (approve, reject-rerun, reject-park, cancel); planning session lifecycle (open/resume/discard/finalize, `QueuePlanningSubtasksAsync`); `RefineTask`, `OpenConPtySessionRequested` (embedded ConPTY terminal), `PickUpInTerminalAsync`; MyDay extras (`IsMyDayList`, `ClearDayCommand`, `ShowPrepLogCommand`) and the pinned Notes pseudo-row (`ShowNotesRow`, `OpenNotesCommand`). Raises `NotesRequested`/`PrepRequested` events consumed by the shell.
|
||||
- **DetailsIslandViewModel** — the detail pane for a bound `TaskRowViewModel`. Owns live-log streaming (`Log` via `StreamLineFormatter`), debounced title/description editing, subtasks, session-outcome/roadblock split (splits `Result` at the roadblock marker into two cards), the three-tab work console (`output`/`git`/`session`), child surfacing (`ChildOutcomes` rows plus `ChildrenNeedingAttention`/`HasChildrenNeedingAttention` — children that failed, were cancelled, await review, or reported roadblocks — drive an attention band on the Session tab, which is only visible when `HasChildOutcomes`), and the modes: `IsNotesMode` (hosts `NotesEditorViewModel`), `IsPrepMode`, computed `IsTaskDetailVisible = !IsNotesMode && !IsPrepMode`. Three concerns are extracted into section VMs exposed as properties: **AgentConfigEditorViewModel** (scope=Task; per-task Model/MaxTurns/AgentPath overrides with `InheritedBadge` + `InheritanceResolver`, additive SystemPrompt, debounced auto-save; exposed as `AgentSettings`), **MergeSectionViewModel** (merge-target selection, mergeability indicator via `MergePreviewPresenter` over `PreviewMergeAsync`, `OpenDiffAsync` and `ReviewCombinedDiffCommand` — both build a `DiffViewerViewModel` and call `ShowDiffViewer`), **PrepPanelViewModel** (daily-prep panel: `PrepLog`, `PlanDayCommand` → `RunDailyPrepNowAsync`, persisted last run via `GetLastPrepLogAsync`). Attachments: `Attachments` (`ObservableCollection<AttachmentRowViewModel>`), `IsDragOver`, `DropStatus`, `CanAcceptDrop`, `AddFilesAsync`, `RemoveAttachmentCommand`; loads on task change; `ComposedPreview` includes attachment paths. Writes directly via `new AttachmentStore()` + `new TaskAttachmentRepository(ctx)`. Helper rows (`ChildOutcomeRowViewModel`, `SubtaskRowViewModel`, `LogLineViewModel`, `AttachmentRowViewModel`) live in the same file.
|
||||
- **TaskRowViewModel** / **ListNavItemViewModel** — lightweight display VMs (task row: status, planning phase, parent/blocked links, roadblock count, computed `IsDraft`/`IsPlanned`/`IsChild`/`IsPlanningParent`/`CanRefine`; list row: kind Smart/Virtual/User, count, icon/dot keys, drop hints).
|
||||
- **NotesEditorViewModel** — day navigator + bullet CRUD for daily notes via `INotesApi`.
|
||||
@@ -45,7 +45,7 @@ Design/ — Tokens.axaml (design tokens; merged before styles) + IslandStyle
|
||||
|
||||
## Services
|
||||
|
||||
- **WorkerClient** / **IWorkerClient** — SignalR client connecting to `http://127.0.0.1:47821/hub`, auto-reconnect with exponential backoff. The surface tracks `WorkerHub` (see `src/ClaudeDo.Worker/CLAUDE.md` for the canonical method/event list); groups: task execution (RunNow/Cancel/Continue/Reset/SetTaskStatus), review (`ApproveReviewAsync(taskId, targetBranch) -> MergeResultDto`, reject-to-queue/idle, cancel review, `PreviewMergeAsync -> MergePreviewDto`), planning sessions (start/resume/discard/finalize, queue subtasks, pending draft count, interactive terminal, refine), planning aggregate/integration-branch diffs, unit-merge continue/abort, single-task conflict resolving (start/get-conflict-documents/write-resolution/continue/abort), worktrees (overview, set state, force remove, cleanup, reset all), agents, app settings, lists/config, weekly report, daily notes, daily prep (`RunDailyPrepNowAsync`, `ClearMyDayAsync`, `GetLastPrepLogAsync`), prime schedules, recent worker logs (`GetRecentLogsAsync`). Events mirror `HubBroadcaster` (task/worktree/list/run updates, prep events, planning-merge events, refine events, worker log). Lifecycle (`StartAsync`/`StopAsync`) and a few admin methods live only on the concrete `WorkerClient`.
|
||||
- **WorkerClient** / **IWorkerClient** — SignalR client connecting to `http://127.0.0.1:47821/hub`, auto-reconnect with exponential backoff. The surface tracks `WorkerHub` (see `src/ClaudeDo.Worker/CLAUDE.md` for the canonical method/event list); groups: task execution (RunNow/Cancel/Continue/Reset/SetTaskStatus), review (`ApproveReviewAsync(taskId, targetBranch) -> MergeResultDto`, reject-to-queue/idle, cancel review, `PreviewMergeAsync -> MergePreviewDto`), planning sessions (start/resume/discard/finalize, queue subtasks, pending draft count, refine), pick-up-in-terminal + embedded ConPTY launch specs (`GetInteractiveLaunchSpecAsync`/`GetAdHocLaunchSpecAsync`), planning aggregate/integration-branch diffs, unit-merge continue/abort, single-task conflict resolving (start/get-conflict-documents/write-resolution/continue/abort), worktrees (overview, set state, force remove, cleanup, reset all), agents, app settings, lists/config, weekly report, daily notes, daily prep (`RunDailyPrepNowAsync`, `ClearMyDayAsync`, `GetLastPrepLogAsync`), prime schedules, recent worker logs (`GetRecentLogsAsync`). Events mirror `HubBroadcaster` (task/worktree/list/run updates, prep events, planning-merge events, refine events, worker log). Lifecycle (`StartAsync`/`StopAsync`) and a few admin methods live only on the concrete `WorkerClient`.
|
||||
- **INotesApi** / **WorkerNotesApi** — daily-note CRUD (`ListAsync(day)`, `AddAsync`, `UpdateAsync`, `DeleteAsync`); UI DTO `DailyNoteDto(Id, Date, Text, SortOrder)`.
|
||||
- **IPrimeScheduleApi** — prime-schedule CRUD (`ListAsync`, `UpsertAsync`, `DeleteAsync`).
|
||||
- **UpdateCheckService** — polls releases, exposes `LastCheckStatus`/`LatestVersion`/`CheckNowAsync` (feeds the shell's update banner).
|
||||
|
||||
@@ -25,11 +25,6 @@ public interface IWorkerClient : INotifyPropertyChanged
|
||||
/// <summary>A pending question was answered, timed out, or the run ended: (taskId, questionId).</summary>
|
||||
event Action<string, string>? TaskQuestionResolvedEvent;
|
||||
|
||||
event Action<string>? InteractiveSessionStartedEvent;
|
||||
event Action<string>? InteractiveSessionEndedEvent;
|
||||
event Action<string, IReadOnlyList<string>>? InteractiveQueueChangedEvent;
|
||||
event Action<string, string>? InteractiveMessageSentEvent;
|
||||
|
||||
event Action? PrepStartedEvent;
|
||||
event Action<string>? PrepLineEvent;
|
||||
event Action<bool>? PrepFinishedEvent;
|
||||
@@ -51,10 +46,6 @@ public interface IWorkerClient : INotifyPropertyChanged
|
||||
Task ContinueTaskAsync(string taskId, string followUpPrompt);
|
||||
/// <summary>Answer a question a running task raised via AskUser.</summary>
|
||||
Task AnswerTaskQuestionAsync(string taskId, string questionId, string answer);
|
||||
Task SendInteractiveMessageAsync(string taskId, string text);
|
||||
Task RemoveQueuedInteractiveMessageAsync(string taskId, string text);
|
||||
Task StopInteractiveSessionAsync(string taskId);
|
||||
Task InterruptInteractiveSessionAsync(string taskId);
|
||||
/// <summary>The question a running task is currently blocked on, if any (for re-attach).</summary>
|
||||
Task<PendingQuestionDto?> GetPendingQuestionAsync(string taskId);
|
||||
Task ResetTaskAsync(string taskId);
|
||||
@@ -83,9 +74,7 @@ public interface IWorkerClient : INotifyPropertyChanged
|
||||
Task<MergeResultDto> ContinueConflictMergeAsync(string taskId);
|
||||
Task AbortConflictMergeAsync(string taskId);
|
||||
Task StartPlanningSessionAsync(string taskId, CancellationToken ct = default);
|
||||
Task OpenInteractiveTerminalAsync(string taskId, CancellationToken ct = default);
|
||||
// Picks up a task's Claude session in a real terminal window (--resume) — distinct
|
||||
// from OpenInteractiveTerminalAsync (the in-app streaming session).
|
||||
// Picks up a task's Claude session in a real terminal window (--resume).
|
||||
Task ResumeTaskInTerminalAsync(string taskId, CancellationToken ct = default);
|
||||
/// <summary>Launch spec for an embedded ConPTY terminal to open an interactive session
|
||||
/// in a task's worktree (same worktree prep as an autonomous run).</summary>
|
||||
|
||||
@@ -49,10 +49,6 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
||||
public event Action<string>? TaskUpdatedEvent;
|
||||
public event Action<string, string, string>? TaskQuestionAskedEvent;
|
||||
public event Action<string, string>? TaskQuestionResolvedEvent;
|
||||
public event Action<string>? InteractiveSessionStartedEvent;
|
||||
public event Action<string>? InteractiveSessionEndedEvent;
|
||||
public event Action<string, IReadOnlyList<string>>? InteractiveQueueChangedEvent;
|
||||
public event Action<string, string>? InteractiveMessageSentEvent;
|
||||
public event Action? ConnectionRestoredEvent;
|
||||
public event Action<string>? WorktreeUpdatedEvent;
|
||||
public event Action<string>? ListUpdatedEvent;
|
||||
@@ -152,26 +148,6 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
||||
Dispatcher.UIThread.Post(() => TaskQuestionResolvedEvent?.Invoke(taskId, questionId));
|
||||
});
|
||||
|
||||
_hub.On<string>("InteractiveSessionStarted", taskId =>
|
||||
{
|
||||
Dispatcher.UIThread.Post(() => InteractiveSessionStartedEvent?.Invoke(taskId));
|
||||
});
|
||||
|
||||
_hub.On<string>("InteractiveSessionEnded", taskId =>
|
||||
{
|
||||
Dispatcher.UIThread.Post(() => InteractiveSessionEndedEvent?.Invoke(taskId));
|
||||
});
|
||||
|
||||
_hub.On<string, IReadOnlyList<string>>("InteractiveQueueChanged", (taskId, pending) =>
|
||||
{
|
||||
Dispatcher.UIThread.Post(() => InteractiveQueueChangedEvent?.Invoke(taskId, pending));
|
||||
});
|
||||
|
||||
_hub.On<string, string>("InteractiveMessageSent", (taskId, text) =>
|
||||
{
|
||||
Dispatcher.UIThread.Post(() => InteractiveMessageSentEvent?.Invoke(taskId, text));
|
||||
});
|
||||
|
||||
_hub.On<string>("WorktreeUpdated", taskId =>
|
||||
{
|
||||
Dispatcher.UIThread.Post(() => WorktreeUpdatedEvent?.Invoke(taskId));
|
||||
@@ -303,30 +279,6 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
||||
catch { /* offline or already resolved — the UI clears optimistically */ }
|
||||
}
|
||||
|
||||
public async Task SendInteractiveMessageAsync(string taskId, string text)
|
||||
{
|
||||
try { await _hub.InvokeAsync("SendInteractiveMessage", taskId, text); }
|
||||
catch { /* offline or session already ended */ }
|
||||
}
|
||||
|
||||
public async Task RemoveQueuedInteractiveMessageAsync(string taskId, string text)
|
||||
{
|
||||
try { await _hub.InvokeAsync("RemoveQueuedInteractiveMessage", taskId, text); }
|
||||
catch { /* offline or session already ended */ }
|
||||
}
|
||||
|
||||
public async Task StopInteractiveSessionAsync(string taskId)
|
||||
{
|
||||
try { await _hub.InvokeAsync("StopInteractiveSession", taskId); }
|
||||
catch { /* offline */ }
|
||||
}
|
||||
|
||||
public async Task InterruptInteractiveSessionAsync(string taskId)
|
||||
{
|
||||
try { await _hub.InvokeAsync("InterruptInteractiveSession", taskId); }
|
||||
catch { /* offline */ }
|
||||
}
|
||||
|
||||
public Task<PendingQuestionDto?> GetPendingQuestionAsync(string taskId)
|
||||
=> TryInvokeAsync<PendingQuestionDto>("GetPendingQuestion", taskId);
|
||||
|
||||
@@ -558,9 +510,6 @@ public partial class WorkerClient : ObservableObject, IAsyncDisposable, IWorkerC
|
||||
public async Task<PlanningSessionResumeInfo> ResumePlanningSessionAsync(string taskId, CancellationToken ct = default)
|
||||
=> await _hub.InvokeAsync<PlanningSessionResumeInfo>("ResumePlanningSessionAsync", taskId, ct);
|
||||
|
||||
public async Task OpenInteractiveTerminalAsync(string taskId, CancellationToken ct = default)
|
||||
=> await _hub.InvokeAsync("OpenInteractiveTerminalAsync", taskId, ct);
|
||||
|
||||
public async Task ResumeTaskInTerminalAsync(string taskId, CancellationToken ct = default)
|
||||
=> await _hub.InvokeAsync("ResumeTaskInTerminal", taskId, ct);
|
||||
|
||||
|
||||
@@ -71,9 +71,6 @@ public sealed partial class TaskMonitorViewModel : ViewModelBase, IMissionContro
|
||||
|
||||
private const string RoadblockMarker = "Roadblocks reported during the run:";
|
||||
|
||||
public ObservableCollection<QueuedMessageViewModel> QueuedMessages { get; } = new();
|
||||
public bool HasQueuedMessages => QueuedMessages.Count > 0;
|
||||
|
||||
// Captured handler delegates for disposal
|
||||
private readonly Action<string, string> _onTaskMessage;
|
||||
private readonly Action<string, string, DateTime> _onTaskStarted;
|
||||
@@ -81,19 +78,6 @@ public sealed partial class TaskMonitorViewModel : ViewModelBase, IMissionContro
|
||||
private readonly Action<string> _onTaskUpdated;
|
||||
private readonly Action<string, string, string> _onTaskQuestionAsked;
|
||||
private readonly Action<string, string> _onTaskQuestionResolved;
|
||||
private readonly Action<string> _onInteractiveStarted;
|
||||
private readonly Action<string> _onInteractiveEnded;
|
||||
private readonly Action<string, IReadOnlyList<string>> _onInteractiveQueueChanged;
|
||||
private readonly Action<string, string> _onInteractiveMessageSent;
|
||||
|
||||
// Interactive composer — active while the worker is in an interactive session.
|
||||
[ObservableProperty]
|
||||
[NotifyCanExecuteChangedFor(nameof(SubmitComposerCommand))]
|
||||
private bool _isInteractiveLive;
|
||||
|
||||
[ObservableProperty]
|
||||
[NotifyCanExecuteChangedFor(nameof(SubmitComposerCommand))]
|
||||
private string _composerDraft = string.Empty;
|
||||
|
||||
// A question the running task raised via AskUser and is blocking on, plus the answer
|
||||
// the user is typing. Ephemeral (in-memory + live events) — the task is still Running.
|
||||
@@ -161,46 +145,6 @@ public sealed partial class TaskMonitorViewModel : ViewModelBase, IMissionContro
|
||||
ClearPendingQuestion();
|
||||
};
|
||||
_worker.TaskQuestionResolvedEvent += _onTaskQuestionResolved;
|
||||
|
||||
_onInteractiveStarted = taskId =>
|
||||
{
|
||||
if (taskId == _subscribedTaskId) { IsInteractiveLive = true; AgentState = "running"; }
|
||||
};
|
||||
_worker.InteractiveSessionStartedEvent += _onInteractiveStarted;
|
||||
|
||||
_onInteractiveEnded = taskId =>
|
||||
{
|
||||
if (taskId != _subscribedTaskId) return;
|
||||
IsInteractiveLive = false;
|
||||
AgentState = "done";
|
||||
QueuedMessages.Clear();
|
||||
OnPropertyChanged(nameof(HasQueuedMessages));
|
||||
};
|
||||
_worker.InteractiveSessionEndedEvent += _onInteractiveEnded;
|
||||
|
||||
_onInteractiveQueueChanged = (taskId, pending) =>
|
||||
{
|
||||
if (taskId != _subscribedTaskId) return;
|
||||
QueuedMessages.Clear();
|
||||
foreach (var m in pending)
|
||||
{
|
||||
var text = m;
|
||||
QueuedMessages.Add(new QueuedMessageViewModel
|
||||
{
|
||||
Text = text,
|
||||
RemoveCommand = new CommunityToolkit.Mvvm.Input.RelayCommand(() => _ = RemoveQueuedAsync(text)),
|
||||
});
|
||||
}
|
||||
OnPropertyChanged(nameof(HasQueuedMessages));
|
||||
};
|
||||
_worker.InteractiveQueueChangedEvent += _onInteractiveQueueChanged;
|
||||
|
||||
_onInteractiveMessageSent = (taskId, text) =>
|
||||
{
|
||||
if (taskId == _subscribedTaskId)
|
||||
Log.Add(new LogLineViewModel { Kind = LogKind.User, Text = text });
|
||||
};
|
||||
_worker.InteractiveMessageSentEvent += _onInteractiveMessageSent;
|
||||
}
|
||||
|
||||
// Surface a pending question (used by live event + re-attach hydration).
|
||||
@@ -210,45 +154,6 @@ public sealed partial class TaskMonitorViewModel : ViewModelBase, IMissionContro
|
||||
PendingQuestion = question;
|
||||
}
|
||||
|
||||
// Used by Mission Control when it creates the monitor after the started event already fired.
|
||||
public void SetInteractiveLive(bool live)
|
||||
{
|
||||
IsInteractiveLive = live;
|
||||
if (live) AgentState = "running";
|
||||
}
|
||||
|
||||
[RelayCommand(CanExecute = nameof(CanSubmitComposer))]
|
||||
private async System.Threading.Tasks.Task SubmitComposer()
|
||||
{
|
||||
if (string.IsNullOrEmpty(_subscribedTaskId)) return;
|
||||
var text = ComposerDraft;
|
||||
if (string.IsNullOrWhiteSpace(text)) return;
|
||||
ComposerDraft = string.Empty;
|
||||
await _worker.SendInteractiveMessageAsync(_subscribedTaskId, text);
|
||||
}
|
||||
|
||||
private bool CanSubmitComposer() => IsInteractiveLive && !string.IsNullOrWhiteSpace(ComposerDraft);
|
||||
|
||||
[RelayCommand]
|
||||
private async System.Threading.Tasks.Task StopInteractive()
|
||||
{
|
||||
if (!string.IsNullOrEmpty(_subscribedTaskId) && IsInteractiveLive)
|
||||
await _worker.StopInteractiveSessionAsync(_subscribedTaskId);
|
||||
}
|
||||
|
||||
[RelayCommand]
|
||||
private async System.Threading.Tasks.Task InterruptInteractive()
|
||||
{
|
||||
if (!string.IsNullOrEmpty(_subscribedTaskId) && IsInteractiveLive)
|
||||
await _worker.InterruptInteractiveSessionAsync(_subscribedTaskId);
|
||||
}
|
||||
|
||||
private async System.Threading.Tasks.Task RemoveQueuedAsync(string text)
|
||||
{
|
||||
if (!string.IsNullOrEmpty(_subscribedTaskId))
|
||||
await _worker.RemoveQueuedInteractiveMessageAsync(_subscribedTaskId, text);
|
||||
}
|
||||
|
||||
private void ClearPendingQuestion()
|
||||
{
|
||||
PendingQuestionId = null;
|
||||
@@ -297,10 +202,6 @@ public sealed partial class TaskMonitorViewModel : ViewModelBase, IMissionContro
|
||||
SessionOutcome = null;
|
||||
Roadblocks = null;
|
||||
ClearPendingQuestion();
|
||||
IsInteractiveLive = false;
|
||||
ComposerDraft = string.Empty;
|
||||
QueuedMessages.Clear();
|
||||
OnPropertyChanged(nameof(HasQueuedMessages));
|
||||
}
|
||||
|
||||
[ObservableProperty]
|
||||
@@ -524,15 +425,5 @@ public sealed partial class TaskMonitorViewModel : ViewModelBase, IMissionContro
|
||||
_worker.TaskUpdatedEvent -= _onTaskUpdated;
|
||||
_worker.TaskQuestionAskedEvent -= _onTaskQuestionAsked;
|
||||
_worker.TaskQuestionResolvedEvent -= _onTaskQuestionResolved;
|
||||
_worker.InteractiveSessionStartedEvent -= _onInteractiveStarted;
|
||||
_worker.InteractiveSessionEndedEvent -= _onInteractiveEnded;
|
||||
_worker.InteractiveQueueChangedEvent -= _onInteractiveQueueChanged;
|
||||
_worker.InteractiveMessageSentEvent -= _onInteractiveMessageSent;
|
||||
}
|
||||
}
|
||||
|
||||
public sealed class QueuedMessageViewModel
|
||||
{
|
||||
public required string Text { get; init; }
|
||||
public required System.Windows.Input.ICommand RemoveCommand { get; init; }
|
||||
}
|
||||
|
||||
@@ -810,18 +810,8 @@ public sealed partial class TasksIslandViewModel : ViewModelBase, IDisposable
|
||||
catch (Exception ex) { ErrorReported?.Invoke(Loc.T("vm.tasksIsland.planningOpenFailed", ex.Message)); }
|
||||
}
|
||||
|
||||
[RelayCommand]
|
||||
private async Task RunInteractivelyAsync(TaskRowViewModel? row)
|
||||
{
|
||||
if (row is null || _worker is null) return;
|
||||
ForegroundHelper.AllowAny();
|
||||
try { await _worker.OpenInteractiveTerminalAsync(row.Id); }
|
||||
catch (Exception ex) { ErrorReported?.Invoke(Loc.T("vm.tasksIsland.runInteractiveFailed", ex.Message)); }
|
||||
}
|
||||
|
||||
// Distinct from RunInteractivelyAsync (the old in-app streaming session) — this opens the
|
||||
// task in an embedded ConPTY terminal pane in the Command Center. The shell owns the
|
||||
// Mission Control view model, so this just raises an event for it to act on.
|
||||
// Opens the task in an embedded ConPTY terminal pane in the Command Center. The shell owns
|
||||
// the Mission Control view model, so this just raises an event for it to act on.
|
||||
public event Action<string>? OpenConPtySessionRequested;
|
||||
|
||||
[RelayCommand]
|
||||
|
||||
@@ -22,7 +22,6 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
private readonly Action<string, string, string, DateTime> _onTaskFinished;
|
||||
private readonly Action<string> _onTaskUpdated;
|
||||
private readonly Action _onConnectionRestored;
|
||||
private readonly Action<string> _onInteractiveStarted;
|
||||
|
||||
public ObservableCollection<TaskMonitorViewModel> Monitors { get; } = new();
|
||||
|
||||
@@ -94,14 +93,6 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
_onConnectionRestored = () => { SeedActive(); _ = RefreshQueueAsync(); };
|
||||
_worker.ConnectionRestoredEvent += _onConnectionRestored;
|
||||
|
||||
_onInteractiveStarted = taskId =>
|
||||
{
|
||||
EnsureMonitor(taskId);
|
||||
var m = Monitors.FirstOrDefault(x => x.SubscribedTaskId == taskId);
|
||||
m?.SetInteractiveLive(true);
|
||||
};
|
||||
_worker.InteractiveSessionStartedEvent += _onInteractiveStarted;
|
||||
|
||||
SeedActive();
|
||||
_ = RefreshQueueAsync();
|
||||
}
|
||||
@@ -231,8 +222,7 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
}
|
||||
|
||||
// Fetches the launch spec for a task's worktree and hosts an embedded ConPTY session as a
|
||||
// Command Center pane (task-based only). A distinct entry point from RunInteractivelyAsync's
|
||||
// streaming session — the two coexist until the streaming stack is removed.
|
||||
// Command Center pane (task-based only).
|
||||
public async System.Threading.Tasks.Task OpenConPtySessionAsync(string taskId)
|
||||
{
|
||||
if (string.IsNullOrEmpty(taskId)) return;
|
||||
@@ -362,7 +352,6 @@ public sealed partial class MissionControlViewModel : ViewModelBase, IDisposable
|
||||
_worker.TaskFinishedEvent -= _onTaskFinished;
|
||||
_worker.TaskUpdatedEvent -= _onTaskUpdated;
|
||||
_worker.ConnectionRestoredEvent -= _onConnectionRestored;
|
||||
_worker.InteractiveSessionStartedEvent -= _onInteractiveStarted;
|
||||
Monitors.CollectionChanged -= OnMonitorsChanged;
|
||||
ConPtySessions.CollectionChanged -= OnConPtySessionsChanged;
|
||||
Panes.CollectionChanged -= OnPanesChanged;
|
||||
|
||||
@@ -263,83 +263,6 @@
|
||||
Command="{Binding RejectReviewCommand}" />
|
||||
</Grid>
|
||||
|
||||
<!-- Interactive composer + queued strip — chat with a live in-app session -->
|
||||
<StackPanel DockPanel.Dock="Bottom" Orientation="Vertical">
|
||||
<!-- Queued messages strip -->
|
||||
<Border IsVisible="{Binding Monitor.HasQueuedMessages}"
|
||||
BorderBrush="{DynamicResource LineBrush}"
|
||||
BorderThickness="0,1,0,0"
|
||||
Padding="12,4">
|
||||
<StackPanel Spacing="2">
|
||||
<TextBlock Classes="meta"
|
||||
Text="{loc:Tr session.composer.queued}"
|
||||
Foreground="{DynamicResource TextMuteBrush}" />
|
||||
<ItemsControl ItemsSource="{Binding Monitor.QueuedMessages}">
|
||||
<ItemsControl.ItemTemplate>
|
||||
<DataTemplate x:DataType="vm:QueuedMessageViewModel">
|
||||
<Grid ColumnDefinitions="Auto,*,Auto" Margin="0,1">
|
||||
<TextBlock Grid.Column="0"
|
||||
Text="⧗"
|
||||
Foreground="{DynamicResource TextMuteBrush}"
|
||||
FontFamily="{StaticResource MonoFont}"
|
||||
FontSize="{StaticResource FontSizeMono}"
|
||||
VerticalAlignment="Center"
|
||||
Margin="0,0,6,0" />
|
||||
<TextBlock Grid.Column="1"
|
||||
Text="{Binding Text}"
|
||||
Foreground="{DynamicResource TextDimBrush}"
|
||||
FontFamily="{StaticResource MonoFont}"
|
||||
FontSize="{StaticResource FontSizeMono}"
|
||||
TextTrimming="CharacterEllipsis"
|
||||
VerticalAlignment="Center" />
|
||||
<Button Grid.Column="2"
|
||||
Classes="title-ctrl"
|
||||
Command="{Binding RemoveCommand}"
|
||||
ToolTip.Tip="{loc:Tr session.composer.unqueue}"
|
||||
Margin="4,0,0,0">
|
||||
<PathIcon Data="{StaticResource Icon.WinClose}" Width="8" Height="8"/>
|
||||
</Button>
|
||||
</Grid>
|
||||
</DataTemplate>
|
||||
</ItemsControl.ItemTemplate>
|
||||
</ItemsControl>
|
||||
</StackPanel>
|
||||
</Border>
|
||||
<!-- Composer input row -->
|
||||
<Grid IsVisible="{Binding Monitor.IsInteractiveLive}"
|
||||
ColumnDefinitions="Auto,*,Auto,Auto"
|
||||
Margin="12,2,12,8">
|
||||
<TextBlock Grid.Column="0" Text="❯"
|
||||
FontFamily="{StaticResource MonoFont}"
|
||||
FontSize="{StaticResource FontSizeMono}"
|
||||
Foreground="{DynamicResource AccentBrush}"
|
||||
VerticalAlignment="Center" Margin="0,0,8,0" />
|
||||
<TextBox Grid.Column="1"
|
||||
Classes="review-prompt"
|
||||
Text="{Binding Monitor.ComposerDraft, Mode=TwoWay, UpdateSourceTrigger=PropertyChanged}"
|
||||
AcceptsReturn="False"
|
||||
TextWrapping="Wrap"
|
||||
MaxHeight="160"
|
||||
PlaceholderText="{loc:Tr session.composer.placeholder}"
|
||||
VerticalContentAlignment="Center"
|
||||
FontFamily="{StaticResource MonoFont}"
|
||||
FontSize="{StaticResource FontSizeMono}">
|
||||
<TextBox.KeyBindings>
|
||||
<KeyBinding Gesture="Enter" Command="{Binding Monitor.SubmitComposerCommand}" />
|
||||
</TextBox.KeyBindings>
|
||||
</TextBox>
|
||||
<Button Grid.Column="2" Classes="prompt-action"
|
||||
VerticalAlignment="Center" Margin="12,0,0,0"
|
||||
Command="{Binding Monitor.InterruptInteractiveCommand}"
|
||||
ToolTip.Tip="{loc:Tr session.composer.interrupt}">
|
||||
<PathIcon Data="{StaticResource Icon.Stop}" Width="10" Height="10"/>
|
||||
</Button>
|
||||
<Button Grid.Column="3" Classes="prompt-action accent" Content="[Send]"
|
||||
VerticalAlignment="Center" Margin="4,0,0,0"
|
||||
Command="{Binding Monitor.SubmitComposerCommand}" />
|
||||
</Grid>
|
||||
</StackPanel>
|
||||
|
||||
<ScrollViewer Name="LogScroll"
|
||||
VerticalScrollBarVisibility="Visible"
|
||||
AllowAutoHide="False"
|
||||
|
||||
@@ -50,79 +50,6 @@
|
||||
</Border>
|
||||
</Grid>
|
||||
|
||||
<!-- ── Queued strip + Composer bar ── -->
|
||||
<StackPanel DockPanel.Dock="Bottom" Orientation="Vertical">
|
||||
<!-- Queued messages strip -->
|
||||
<Border IsVisible="{Binding #Root.HasQueuedMessages}"
|
||||
Background="{DynamicResource Surface2Brush}"
|
||||
BorderBrush="{DynamicResource LineBrush}"
|
||||
BorderThickness="0,0,0,1"
|
||||
Padding="8,4">
|
||||
<StackPanel Spacing="2">
|
||||
<TextBlock Classes="meta"
|
||||
Text="{loc:Tr session.composer.queued}"
|
||||
Foreground="{DynamicResource TextMuteBrush}" />
|
||||
<ItemsControl ItemsSource="{Binding #Root.QueuedMessages}">
|
||||
<ItemsControl.ItemTemplate>
|
||||
<DataTemplate x:DataType="vm:QueuedMessageViewModel">
|
||||
<Grid ColumnDefinitions="Auto,*,Auto" Margin="0,1">
|
||||
<TextBlock Grid.Column="0"
|
||||
Text="⧗"
|
||||
Foreground="{DynamicResource TextMuteBrush}"
|
||||
FontFamily="{StaticResource MonoFont}"
|
||||
FontSize="{StaticResource FontSizeMono}"
|
||||
VerticalAlignment="Center"
|
||||
Margin="0,0,6,0" />
|
||||
<TextBlock Grid.Column="1"
|
||||
Text="{Binding Text}"
|
||||
Foreground="{DynamicResource TextDimBrush}"
|
||||
FontFamily="{StaticResource MonoFont}"
|
||||
FontSize="{StaticResource FontSizeMono}"
|
||||
TextTrimming="CharacterEllipsis"
|
||||
VerticalAlignment="Center" />
|
||||
<Button Grid.Column="2"
|
||||
Classes="title-ctrl"
|
||||
Command="{Binding RemoveCommand}"
|
||||
ToolTip.Tip="{loc:Tr session.composer.unqueue}"
|
||||
Margin="4,0,0,0">
|
||||
<PathIcon Data="{StaticResource Icon.WinClose}" Width="8" Height="8"/>
|
||||
</Button>
|
||||
</Grid>
|
||||
</DataTemplate>
|
||||
</ItemsControl.ItemTemplate>
|
||||
</ItemsControl>
|
||||
</StackPanel>
|
||||
</Border>
|
||||
<!-- Composer input row -->
|
||||
<Border IsVisible="{Binding #Root.IsComposerVisible}"
|
||||
Background="{DynamicResource Surface2Brush}"
|
||||
BorderBrush="{DynamicResource LineBrush}"
|
||||
BorderThickness="0,1,0,0"
|
||||
Padding="6,5">
|
||||
<Grid ColumnDefinitions="*,Auto,Auto">
|
||||
<TextBox Grid.Column="0"
|
||||
Text="{Binding #Root.ComposerText, Mode=TwoWay, UpdateSourceTrigger=PropertyChanged}"
|
||||
PlaceholderText="{Binding #Root.ComposerPlaceholder}"
|
||||
AcceptsReturn="False">
|
||||
<TextBox.KeyBindings>
|
||||
<KeyBinding Gesture="Enter" Command="{Binding #Root.SubmitCommand}"/>
|
||||
</TextBox.KeyBindings>
|
||||
</TextBox>
|
||||
<Button Grid.Column="1"
|
||||
Margin="6,0,0,0"
|
||||
Classes="title-ctrl"
|
||||
Command="{Binding #Root.InterruptCommand}"
|
||||
ToolTip.Tip="{loc:Tr session.composer.interrupt}">
|
||||
<PathIcon Data="{StaticResource Icon.Stop}" Width="10" Height="10"/>
|
||||
</Button>
|
||||
<Button Grid.Column="2"
|
||||
Margin="6,0,0,0"
|
||||
Content="{loc:Tr session.composer.send}"
|
||||
Command="{Binding #Root.SubmitCommand}"/>
|
||||
</Grid>
|
||||
</Border>
|
||||
</StackPanel>
|
||||
|
||||
<!-- ── Log output ── -->
|
||||
<ScrollViewer Name="LogScroll"
|
||||
VerticalScrollBarVisibility="Visible"
|
||||
|
||||
@@ -1,9 +1,7 @@
|
||||
using System.Collections;
|
||||
using System.Collections.Specialized;
|
||||
using System.Windows.Input;
|
||||
using Avalonia;
|
||||
using Avalonia.Controls;
|
||||
using Avalonia.Data;
|
||||
|
||||
namespace ClaudeDo.Ui.Views.Islands;
|
||||
|
||||
@@ -19,33 +17,12 @@ public partial class SessionTerminalView : UserControl
|
||||
AvaloniaProperty.Register<SessionTerminalView, bool>(nameof(IsDone));
|
||||
public static readonly StyledProperty<bool> IsFailedProperty =
|
||||
AvaloniaProperty.Register<SessionTerminalView, bool>(nameof(IsFailed));
|
||||
public static readonly StyledProperty<bool> IsComposerVisibleProperty =
|
||||
AvaloniaProperty.Register<SessionTerminalView, bool>(nameof(IsComposerVisible), defaultValue: false);
|
||||
public static readonly StyledProperty<string?> ComposerTextProperty =
|
||||
AvaloniaProperty.Register<SessionTerminalView, string?>(nameof(ComposerText), defaultBindingMode: BindingMode.TwoWay);
|
||||
public static readonly StyledProperty<ICommand?> SubmitCommandProperty =
|
||||
AvaloniaProperty.Register<SessionTerminalView, ICommand?>(nameof(SubmitCommand));
|
||||
public static readonly StyledProperty<ICommand?> InterruptCommandProperty =
|
||||
AvaloniaProperty.Register<SessionTerminalView, ICommand?>(nameof(InterruptCommand));
|
||||
public static readonly StyledProperty<string?> ComposerPlaceholderProperty =
|
||||
AvaloniaProperty.Register<SessionTerminalView, string?>(nameof(ComposerPlaceholder));
|
||||
public static readonly StyledProperty<System.Collections.IEnumerable?> QueuedMessagesProperty =
|
||||
AvaloniaProperty.Register<SessionTerminalView, System.Collections.IEnumerable?>(nameof(QueuedMessages));
|
||||
public static readonly StyledProperty<bool> HasQueuedMessagesProperty =
|
||||
AvaloniaProperty.Register<SessionTerminalView, bool>(nameof(HasQueuedMessages), defaultValue: false);
|
||||
|
||||
public IEnumerable? Entries { get => GetValue(EntriesProperty); set => SetValue(EntriesProperty, value); }
|
||||
public string? Label { get => GetValue(LabelProperty); set => SetValue(LabelProperty, value); }
|
||||
public bool IsRunning { get => GetValue(IsRunningProperty); set => SetValue(IsRunningProperty, value); }
|
||||
public bool IsDone { get => GetValue(IsDoneProperty); set => SetValue(IsDoneProperty, value); }
|
||||
public bool IsFailed { get => GetValue(IsFailedProperty); set => SetValue(IsFailedProperty, value); }
|
||||
public bool IsComposerVisible { get => GetValue(IsComposerVisibleProperty); set => SetValue(IsComposerVisibleProperty, value); }
|
||||
public string? ComposerText { get => GetValue(ComposerTextProperty); set => SetValue(ComposerTextProperty, value); }
|
||||
public ICommand? SubmitCommand { get => GetValue(SubmitCommandProperty); set => SetValue(SubmitCommandProperty, value); }
|
||||
public ICommand? InterruptCommand { get => GetValue(InterruptCommandProperty); set => SetValue(InterruptCommandProperty, value); }
|
||||
public string? ComposerPlaceholder { get => GetValue(ComposerPlaceholderProperty); set => SetValue(ComposerPlaceholderProperty, value); }
|
||||
public System.Collections.IEnumerable? QueuedMessages { get => GetValue(QueuedMessagesProperty); set => SetValue(QueuedMessagesProperty, value); }
|
||||
public bool HasQueuedMessages { get => GetValue(HasQueuedMessagesProperty); set => SetValue(HasQueuedMessagesProperty, value); }
|
||||
|
||||
private INotifyCollectionChanged? _subscribedCollection;
|
||||
|
||||
|
||||
@@ -49,8 +49,6 @@
|
||||
<MenuItem Header="{loc:Tr tasks.ctxMarkCancelled}" Tag="Cancelled" Click="OnSetStatusClick"/>
|
||||
</MenuItem>
|
||||
<Separator/>
|
||||
<MenuItem Header="{loc:Tr tasks.ctxRunInteractively}"
|
||||
Click="OnRunInteractivelyClick"/>
|
||||
<MenuItem Header="{loc:Tr tasks.ctxOpenConPtySession}"
|
||||
Click="OnOpenConPtySessionClick"/>
|
||||
<MenuItem Header="{loc:Tr tasks.ctxPickUpInTerminal}"
|
||||
|
||||
@@ -61,12 +61,6 @@ public partial class TaskRowView : UserControl
|
||||
await vm.OpenPlanningSessionCommand.ExecuteAsync(row);
|
||||
}
|
||||
|
||||
private async void OnRunInteractivelyClick(object? sender, RoutedEventArgs e)
|
||||
{
|
||||
if (DataContext is TaskRowViewModel row && FindTasksVm() is { } vm)
|
||||
await vm.RunInteractivelyCommand.ExecuteAsync(row);
|
||||
}
|
||||
|
||||
private void OnOpenConPtySessionClick(object? sender, RoutedEventArgs e)
|
||||
{
|
||||
if (DataContext is TaskRowViewModel row && FindTasksVm() is { } vm)
|
||||
|
||||
@@ -107,14 +107,7 @@
|
||||
Label="{Binding DisplayTitle}"
|
||||
IsRunning="{Binding IsRunning}"
|
||||
IsDone="{Binding IsDone}"
|
||||
IsFailed="{Binding IsFailed}"
|
||||
IsComposerVisible="{Binding IsInteractiveLive}"
|
||||
ComposerText="{Binding ComposerDraft, Mode=TwoWay}"
|
||||
SubmitCommand="{Binding SubmitComposerCommand}"
|
||||
InterruptCommand="{Binding InterruptInteractiveCommand}"
|
||||
ComposerPlaceholder="{loc:Tr session.composer.placeholder}"
|
||||
QueuedMessages="{Binding QueuedMessages}"
|
||||
HasQueuedMessages="{Binding HasQueuedMessages}" />
|
||||
IsFailed="{Binding IsFailed}" />
|
||||
|
||||
</DockPanel>
|
||||
</Border>
|
||||
|
||||
@@ -12,7 +12,7 @@ Worker/
|
||||
Worktrees/ — WorktreeMaintenanceService
|
||||
Agents/ — AgentFileService, DefaultAgentSeeder
|
||||
Runner/ — TaskRunner + Claude CLI integration; TaskRunMcpService/TaskRunMcpContext/TaskRunTokenRegistry (in-task MCP wired during execution)
|
||||
Planning/ — PlanningSessionManager, PlanningChainCoordinator, PlanningMcpService, PlanningMergeOrchestrator, PlanningAggregator, PlanningSessionContext/PlanningTokenAuth/PlanningMcpContextAccessor, WindowsTerminalLauncher (ITerminalLauncher) — wt launcher for planning + interactive sessions
|
||||
Planning/ — PlanningSessionManager, PlanningChainCoordinator, PlanningMcpService, PlanningMergeOrchestrator, PlanningAggregator, PlanningSessionContext/PlanningTokenAuth/PlanningMcpContextAccessor, WindowsTerminalLauncher (ITerminalLauncher) — wt launcher for planning sessions + pick-up-in-terminal
|
||||
Refine/ — RefineRunner + RefinePrompt (hub `RefineTask`; broadcasts RefineStarted/RefineFinished)
|
||||
External/ — ExternalMcpService + sibling tool classes
|
||||
Config/ — WorkerConfig
|
||||
@@ -154,7 +154,8 @@ Each CLI invocation is recorded in the `task_runs` table via `TaskRunRepository`
|
||||
- Execution: `Ping`, `GetActive`, `RunNow`, `CancelTask`, `WakeQueue`, `ContinueTask`, `ResetTask`, `SetTaskStatus`, `RefineTask`
|
||||
- Review/merge: `ApproveReview(taskId, targetBranch) -> MergeResultDto` (childless task: merges its worktree then Done, conflict stays WaitingForReview; task with children: drives `PlanningMergeOrchestrator` to merge the whole unit), `ContinuePlanningMerge` / `AbortPlanningMerge` (resolve a unit-merge conflict), `PreviewMerge(taskId, targetBranch) -> MergePreviewDto` (non-destructive mergeability check), `RejectReviewToQueue`, `RejectReviewToIdle`, `CancelReview`, `MergeTask`, `GetMergeTargets`
|
||||
- Single-task conflict resolver (Layer C): `StartConflictMerge`, `GetMergeConflictDocuments` (segments), `WriteConflictResolution`, `ContinueConflictMerge`, `AbortConflictMerge` (service-level `TaskMergeService.ContinueMergeAsync`/`AbortMergeAsync` keep their names)
|
||||
- Planning sessions: `StartPlanningSession`, `ResumePlanningSession`, `DiscardPlanningSession`, `FinalizePlanningSession`, `QueuePlanningSubtasks`, `GetPendingDraftCount`, `OpenInteractiveTerminal`, `GetPlanningAggregate` (per-subtask diffs), `BuildPlanningIntegrationBranch` (combined diff)
|
||||
- Planning sessions: `StartPlanningSession`, `ResumePlanningSession`, `DiscardPlanningSession`, `FinalizePlanningSession`, `QueuePlanningSubtasks`, `GetPendingDraftCount`, `GetPlanningAggregate` (per-subtask diffs), `BuildPlanningIntegrationBranch` (combined diff)
|
||||
- Interactive sessions (embedded ConPTY, UI process): `ResumeTaskInTerminal` (pick-up-in-terminal), `GetInteractiveLaunchSpec`, `GetAdHocLaunchSpec`
|
||||
- Worktrees: `CleanupFinishedWorktrees`, `ResetAllWorktrees`, `GetWorktreesOverview`, `SetWorktreeState`, `ForceRemoveWorktree`
|
||||
- Agents/settings/lists: `GetAgents`, `RefreshAgents`, `RestoreDefaultAgents`, `GetAppSettings`, `UpdateAppSettings`, `UpdateList`, `UpdateListConfig`, `GetListConfig`, `UpdateTaskAgentSettings`
|
||||
- Reports/notes/prep: `GetWeekReport`, `GenerateWeekReport`, `GetDailyNotes`, `AddDailyNote`, `UpdateDailyNote`, `DeleteDailyNote`, `RunDailyPrepNow`, `ClearMyDay`, `GetLastPrepLog`, `ListPrimeSchedules`, `UpsertPrimeSchedule`, `DeletePrimeSchedule`
|
||||
|
||||
@@ -41,10 +41,6 @@ public sealed class WorkerConfig
|
||||
[JsonPropertyName("external_mcp_api_key")]
|
||||
public string? ExternalMcpApiKey { get; set; }
|
||||
|
||||
/// <summary>Interactive/streaming sessions idle longer than this are stopped by IdleSessionReaper. 0 disables reaping.</summary>
|
||||
[JsonPropertyName("interactive_idle_timeout_minutes")]
|
||||
public int InteractiveIdleTimeoutMinutes { get; set; } = 30;
|
||||
|
||||
[JsonPropertyName("online_inbox")]
|
||||
public OnlineInboxConfig OnlineInbox { get; set; } = new();
|
||||
|
||||
|
||||
@@ -77,16 +77,4 @@ public sealed class HubBroadcaster : IPrimeBroadcaster, IRefineBroadcaster
|
||||
Task IRefineBroadcaster.RefineStartedAsync(string taskId) => RefineStarted(taskId);
|
||||
Task IRefineBroadcaster.RefineFinishedAsync(string taskId, bool success, string? error) =>
|
||||
RefineFinished(taskId, success, error);
|
||||
|
||||
public Task InteractiveSessionStarted(string taskId) =>
|
||||
_hub.Clients.All.SendAsync("InteractiveSessionStarted", taskId);
|
||||
|
||||
public Task InteractiveSessionEnded(string taskId) =>
|
||||
_hub.Clients.All.SendAsync("InteractiveSessionEnded", taskId);
|
||||
|
||||
public Task InteractiveQueueChanged(string taskId, IReadOnlyList<string> pending) =>
|
||||
_hub.Clients.All.SendAsync("InteractiveQueueChanged", taskId, pending);
|
||||
|
||||
public Task InteractiveMessageSent(string taskId, string text) =>
|
||||
_hub.Clients.All.SendAsync("InteractiveMessageSent", taskId, text);
|
||||
}
|
||||
|
||||
@@ -127,7 +127,6 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
private readonly OnlineInboxConfig _onlineInboxConfig;
|
||||
private readonly OnlineTokenStore _onlineTokenStore;
|
||||
private readonly Runner.PendingQuestionRegistry _pendingQuestions;
|
||||
private readonly InteractiveSessionService _interactive;
|
||||
private readonly LogRingBuffer? _logBuffer;
|
||||
private readonly ISessionSkillRegistry _skillRegistry;
|
||||
private readonly IInteractiveLaunchSpecService? _interactiveLaunchSpec;
|
||||
@@ -156,7 +155,6 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
OnlineInboxConfig onlineInboxConfig,
|
||||
OnlineTokenStore onlineTokenStore,
|
||||
Runner.PendingQuestionRegistry pendingQuestions,
|
||||
InteractiveSessionService interactive,
|
||||
ISessionSkillRegistry skillRegistry,
|
||||
LogRingBuffer? logBuffer = null,
|
||||
IInteractiveLaunchSpecService? interactiveLaunchSpec = null)
|
||||
@@ -184,7 +182,6 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
_onlineInboxConfig = onlineInboxConfig;
|
||||
_onlineTokenStore = onlineTokenStore;
|
||||
_pendingQuestions = pendingQuestions;
|
||||
_interactive = interactive;
|
||||
_skillRegistry = skillRegistry;
|
||||
_logBuffer = logBuffer;
|
||||
_interactiveLaunchSpec = interactiveLaunchSpec;
|
||||
@@ -624,13 +621,9 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
return ctx;
|
||||
}
|
||||
|
||||
public Task OpenInteractiveTerminalAsync(string taskId) =>
|
||||
_interactive.StartAsync(taskId, Context.ConnectionAborted);
|
||||
|
||||
// Picks up a task's Claude session in a real terminal window (--resume) so the user can
|
||||
// drive it by hand — distinct from OpenInteractiveTerminalAsync (the in-app streaming
|
||||
// session). Only for tasks the worker isn't actively running, with a persisted session
|
||||
// id and a live worktree.
|
||||
// drive it by hand. Only for tasks the worker isn't actively running, with a persisted
|
||||
// session id and a live worktree.
|
||||
public Task ResumeTaskInTerminal(string taskId) => HubGuard(async () =>
|
||||
{
|
||||
await using var ctx = await _dbFactory.CreateDbContextAsync();
|
||||
@@ -680,18 +673,6 @@ public sealed class WorkerHub : Microsoft.AspNetCore.SignalR.Hub
|
||||
return _interactiveLaunchSpec.BuildForDirectoryAsync(directory, Context.ConnectionAborted);
|
||||
});
|
||||
|
||||
public Task SendInteractiveMessage(string taskId, string text) =>
|
||||
_interactive.SendAsync(taskId, text, Context.ConnectionAborted);
|
||||
|
||||
public Task StopInteractiveSession(string taskId) =>
|
||||
_interactive.StopAsync(taskId, Context.ConnectionAborted);
|
||||
|
||||
public Task InterruptInteractiveSession(string taskId) =>
|
||||
_interactive.InterruptAsync(taskId, Context.ConnectionAborted);
|
||||
|
||||
public Task RemoveQueuedInteractiveMessage(string taskId, string text) =>
|
||||
_interactive.RemoveQueuedAsync(taskId, text, Context.ConnectionAborted);
|
||||
|
||||
public async Task<DiscardPlanningOutcome> DiscardPlanningSessionAsync(string taskId, bool dequeueQueuedChildren = false)
|
||||
{
|
||||
var outcome = await _planning.DiscardAsync(taskId, dequeueQueuedChildren, Context.ConnectionAborted);
|
||||
|
||||
@@ -1,172 +0,0 @@
|
||||
using System.Text;
|
||||
using ClaudeDo.Data;
|
||||
using ClaudeDo.Data.Models;
|
||||
using ClaudeDo.Data.Repositories;
|
||||
using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Hub;
|
||||
using ClaudeDo.Worker.Runner;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
namespace ClaudeDo.Worker.Planning;
|
||||
|
||||
public sealed class InteractiveSessionService
|
||||
{
|
||||
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
|
||||
private readonly WorkerConfig _cfg;
|
||||
private readonly HubBroadcaster _broadcaster;
|
||||
private readonly LiveSessionRegistry _registry;
|
||||
private readonly ILoggerFactory _loggerFactory;
|
||||
|
||||
// Optional factory for tests. Signature: (onLine) -> (session, waitForExitTask).
|
||||
// The waitForExitTask completes when the underlying process has exited.
|
||||
private readonly Func<string, IReadOnlyList<string>, Func<string, Task>, (ILiveSession session, Task exitTask)>? _sessionFactory;
|
||||
|
||||
public InteractiveSessionService(
|
||||
IDbContextFactory<ClaudeDoDbContext> dbFactory,
|
||||
WorkerConfig cfg,
|
||||
HubBroadcaster broadcaster,
|
||||
LiveSessionRegistry registry,
|
||||
ILoggerFactory loggerFactory,
|
||||
Func<string, IReadOnlyList<string>, Func<string, Task>, (ILiveSession session, Task exitTask)>? sessionFactory = null)
|
||||
{
|
||||
_dbFactory = dbFactory;
|
||||
_cfg = cfg;
|
||||
_broadcaster = broadcaster;
|
||||
_registry = registry;
|
||||
_loggerFactory = loggerFactory;
|
||||
_sessionFactory = sessionFactory;
|
||||
}
|
||||
|
||||
public async Task StartAsync(string taskId, CancellationToken ct)
|
||||
{
|
||||
if (_registry.TryGet(taskId, out _))
|
||||
throw new InvalidOperationException("An interactive session is already running for this task.");
|
||||
|
||||
await using var ctx = _dbFactory.CreateDbContext();
|
||||
var tasks = new TaskRepository(ctx);
|
||||
var lists = new ListRepository(ctx);
|
||||
|
||||
var task = await tasks.GetByIdAsync(taskId, ct)
|
||||
?? throw new InvalidOperationException($"Task {taskId} not found.");
|
||||
var list = await lists.GetByIdAsync(task.ListId, ct)
|
||||
?? throw new InvalidOperationException($"List {task.ListId} not found.");
|
||||
|
||||
var workingDir = list.WorkingDir;
|
||||
if (string.IsNullOrWhiteSpace(workingDir) || !Directory.Exists(workingDir))
|
||||
throw new InvalidOperationException(
|
||||
$"List '{list.Name}' has no valid working directory configured.");
|
||||
|
||||
var seededPrompt = BuildInteractivePrompt(task);
|
||||
|
||||
var args = new[]
|
||||
{
|
||||
"-p",
|
||||
"--input-format", "stream-json",
|
||||
"--output-format", "stream-json",
|
||||
"--verbose",
|
||||
"--replay-user-messages",
|
||||
"--model", ModelRegistry.PlanningAlias,
|
||||
"--permission-mode", "auto",
|
||||
};
|
||||
|
||||
Func<string, Task> onLine = line =>
|
||||
{
|
||||
_registry.Touch(taskId);
|
||||
return _broadcaster.TaskMessage(taskId, "[stdout] " + line);
|
||||
};
|
||||
|
||||
ILiveSession session;
|
||||
Task exitTask;
|
||||
|
||||
if (_sessionFactory is not null)
|
||||
{
|
||||
// Factory is responsible for providing a ready-to-use session and its exit signal.
|
||||
(session, exitTask) = _sessionFactory(workingDir, args, onLine);
|
||||
}
|
||||
else
|
||||
{
|
||||
var transport = new ProcessClaudeStreamTransport(
|
||||
_cfg,
|
||||
_loggerFactory.CreateLogger<ProcessClaudeStreamTransport>());
|
||||
var streamingSession = new StreamingClaudeSession(
|
||||
transport,
|
||||
onLine,
|
||||
_loggerFactory.CreateLogger<StreamingClaudeSession>(),
|
||||
onQueueChanged: pending => _ = _broadcaster.InteractiveQueueChanged(taskId, pending),
|
||||
onUserMessageSent: text => _ = _broadcaster.InteractiveMessageSent(taskId, text));
|
||||
await streamingSession.StartAsync(args, workingDir, seededPrompt, ct);
|
||||
session = streamingSession;
|
||||
exitTask = transport.WaitForExitAsync();
|
||||
}
|
||||
_registry.Register(taskId, session);
|
||||
await _broadcaster.InteractiveSessionStarted(taskId);
|
||||
|
||||
var logger = _loggerFactory.CreateLogger<InteractiveSessionService>();
|
||||
_ = WatchExitAsync(taskId, exitTask, logger);
|
||||
}
|
||||
|
||||
private async Task WatchExitAsync(string taskId, Task exitTask, ILogger logger)
|
||||
{
|
||||
try
|
||||
{
|
||||
await exitTask;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogWarning(ex, "Interactive session exit watcher caught an exception for task {task_id}", taskId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_registry.Unregister(taskId);
|
||||
try { await _broadcaster.InteractiveSessionEnded(taskId); }
|
||||
catch (Exception ex) { logger.LogWarning(ex, "InteractiveSessionEnded broadcast failed for task {task_id}", taskId); }
|
||||
}
|
||||
}
|
||||
|
||||
public async Task SendAsync(string taskId, string text, CancellationToken ct)
|
||||
{
|
||||
if (!_registry.TryGet(taskId, out var session))
|
||||
throw new InvalidOperationException("No interactive session is running for this task.");
|
||||
_registry.Touch(taskId);
|
||||
await session.SendUserMessageAsync(text, ct);
|
||||
}
|
||||
|
||||
public async Task RemoveQueuedAsync(string taskId, string text, CancellationToken ct)
|
||||
{
|
||||
if (_registry.TryGet(taskId, out var session))
|
||||
{
|
||||
_registry.Touch(taskId);
|
||||
await session.RemoveQueuedAsync(text, ct);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task InterruptAsync(string taskId, CancellationToken ct)
|
||||
{
|
||||
if (_registry.TryGet(taskId, out var session))
|
||||
{
|
||||
_registry.Touch(taskId);
|
||||
await session.InterruptAsync(ct);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task StopAsync(string taskId, CancellationToken ct)
|
||||
{
|
||||
// StopAsync removes from registry and kills the session.
|
||||
// The exit watcher will fire InteractiveSessionEnded once the process exits,
|
||||
// so we don't broadcast here — the watcher is the single authoritative source.
|
||||
await _registry.StopAsync(taskId);
|
||||
}
|
||||
|
||||
private static string BuildInteractivePrompt(TaskEntity task)
|
||||
{
|
||||
var sb = new StringBuilder();
|
||||
sb.AppendLine($"# Task: {task.Title}");
|
||||
if (!string.IsNullOrWhiteSpace(task.Description))
|
||||
{
|
||||
sb.AppendLine();
|
||||
sb.AppendLine(task.Description);
|
||||
}
|
||||
return sb.ToString();
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,7 @@ namespace ClaudeDo.Worker.Planning;
|
||||
|
||||
// Launches the Claude CLI in a visible terminal for human-driven planning sessions.
|
||||
// Not used for headless task execution (that path is ClaudeProcess, prompt over stdin)
|
||||
// nor for interactive sessions (those use InteractiveSessionService + StreamingClaudeSession).
|
||||
// nor for embedded ConPTY interactive sessions (those use IInteractiveLaunchSpecService).
|
||||
public interface ITerminalLauncher
|
||||
{
|
||||
Task LaunchPlanningStartAsync(PlanningSessionStartContext ctx, CancellationToken cancellationToken);
|
||||
|
||||
@@ -85,9 +85,6 @@ builder.Services.AddSingleton<TaskMergeService>();
|
||||
builder.Services.AddSingleton<PlanningAggregator>();
|
||||
builder.Services.AddSingleton<PlanningMergeOrchestrator>();
|
||||
builder.Services.AddSingleton<PlanningChainCoordinator>();
|
||||
builder.Services.AddSingleton<LiveSessionRegistry>();
|
||||
builder.Services.AddSingleton<InteractiveSessionService>();
|
||||
builder.Services.AddHostedService<IdleSessionReaper>();
|
||||
|
||||
// Queue dispatch primitives. QueueWaker holds the wake semaphore; the queue picker
|
||||
// performs atomic Queued→Running claim. Both injected into the state service so it
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
using ClaudeDo.Worker.Config;
|
||||
|
||||
namespace ClaudeDo.Worker.Runner;
|
||||
|
||||
// Stops interactive/streaming sessions that have gone idle. Interactive `claude` processes wait
|
||||
// on stdin and never exit on their own, and there is no client-disconnect teardown — so an
|
||||
// abandoned chat (UI closed, navigated away, crashed) keeps its claude.exe (+ conhost) alive for
|
||||
// the worker's entire lifetime. Under a long-running autostart worker these pile up (observed:
|
||||
// ~170 child processes). This sweep reaps the idle ones.
|
||||
public sealed class IdleSessionReaper : BackgroundService
|
||||
{
|
||||
private static readonly TimeSpan SweepInterval = TimeSpan.FromMinutes(5);
|
||||
|
||||
private readonly LiveSessionRegistry _registry;
|
||||
private readonly WorkerConfig _cfg;
|
||||
private readonly ILogger<IdleSessionReaper> _logger;
|
||||
|
||||
public IdleSessionReaper(LiveSessionRegistry registry, WorkerConfig cfg, ILogger<IdleSessionReaper> logger)
|
||||
{
|
||||
_registry = registry;
|
||||
_cfg = cfg;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
var idleTimeout = TimeSpan.FromMinutes(_cfg.InteractiveIdleTimeoutMinutes);
|
||||
if (idleTimeout <= TimeSpan.Zero)
|
||||
return; // reaper disabled
|
||||
|
||||
using var timer = new PeriodicTimer(SweepInterval);
|
||||
while (await timer.WaitForNextTickAsync(stoppingToken))
|
||||
{
|
||||
try
|
||||
{
|
||||
var reaped = await _registry.ReapIdleAsync(DateTime.UtcNow, idleTimeout);
|
||||
if (reaped.Count > 0)
|
||||
_logger.LogInformation(
|
||||
"Reaped {session_count} idle interactive session(s) after {idle_minutes} min: {task_ids}",
|
||||
reaped.Count, _cfg.InteractiveIdleTimeoutMinutes, string.Join(", ", reaped));
|
||||
}
|
||||
catch (OperationCanceledException) { throw; }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Idle session reap sweep failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,11 +0,0 @@
|
||||
namespace ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
public interface IClaudeStreamTransport : IAsyncDisposable
|
||||
{
|
||||
Task StartAsync(IReadOnlyList<string> args, string workingDirectory, CancellationToken ct);
|
||||
Task WriteLineAsync(string jsonLine, CancellationToken ct);
|
||||
event Func<string, Task>? LineReceived;
|
||||
event Func<string, Task>? StderrReceived;
|
||||
void Kill();
|
||||
Task WaitForExitAsync();
|
||||
}
|
||||
@@ -1,10 +0,0 @@
|
||||
namespace ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
public interface ILiveSession : IAsyncDisposable
|
||||
{
|
||||
bool IsTurnInFlight { get; }
|
||||
Task SendUserMessageAsync(string text, CancellationToken ct);
|
||||
Task RemoveQueuedAsync(string text, CancellationToken ct);
|
||||
Task InterruptAsync(CancellationToken ct);
|
||||
Task StopAsync();
|
||||
}
|
||||
@@ -1,87 +0,0 @@
|
||||
using System.Collections.Concurrent;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
namespace ClaudeDo.Worker.Runner;
|
||||
|
||||
// Singleton in-memory registry of active live streaming sessions.
|
||||
// A session's lifetime matches its associated task run; dead entries are removed by the runner.
|
||||
//
|
||||
// Interactive (stream-json) sessions never exit on their own — they wait on stdin — and there is
|
||||
// no client-disconnect teardown, so an abandoned chat would otherwise keep its claude.exe alive
|
||||
// for the worker's whole lifetime. IdleSessionReaper periodically stops sessions that have seen
|
||||
// no activity past a timeout (see ReapIdleAsync); Touch() records that activity.
|
||||
public sealed class LiveSessionRegistry
|
||||
{
|
||||
private sealed class Entry
|
||||
{
|
||||
public required ILiveSession Session { get; init; }
|
||||
public long LastActivityTicksUtc;
|
||||
}
|
||||
|
||||
private readonly ConcurrentDictionary<string, Entry> _sessions = new();
|
||||
|
||||
public void Register(string taskId, ILiveSession session)
|
||||
{
|
||||
if (_sessions.TryRemove(taskId, out var existing))
|
||||
{
|
||||
// Best-effort stop of the replaced session; don't await to avoid deadlock risk.
|
||||
_ = existing.Session.StopAsync().ContinueWith(t =>
|
||||
{
|
||||
if (t.IsFaulted) { /* swallow — old session is already orphaned */ }
|
||||
}, TaskScheduler.Default);
|
||||
}
|
||||
_sessions[taskId] = new Entry { Session = session, LastActivityTicksUtc = DateTime.UtcNow.Ticks };
|
||||
}
|
||||
|
||||
// Marks a session as active so the idle reaper leaves it alone. Called on every user
|
||||
// message and every output line. No-op if the session is not (yet) registered.
|
||||
public void Touch(string taskId)
|
||||
{
|
||||
if (_sessions.TryGetValue(taskId, out var entry))
|
||||
Interlocked.Exchange(ref entry.LastActivityTicksUtc, DateTime.UtcNow.Ticks);
|
||||
}
|
||||
|
||||
public bool TryGet(string taskId, out ILiveSession session)
|
||||
{
|
||||
if (_sessions.TryGetValue(taskId, out var entry))
|
||||
{
|
||||
session = entry.Session;
|
||||
return true;
|
||||
}
|
||||
session = null!;
|
||||
return false;
|
||||
}
|
||||
|
||||
public void Unregister(string taskId) => _sessions.TryRemove(taskId, out _);
|
||||
|
||||
public async Task StopAsync(string taskId)
|
||||
{
|
||||
if (_sessions.TryRemove(taskId, out var entry))
|
||||
await entry.Session.StopAsync();
|
||||
}
|
||||
|
||||
// Stops and removes every session whose last activity is older than (nowUtc - idleTimeout),
|
||||
// skipping any session with a turn in flight (an agent that's actively working, even if quiet).
|
||||
// Returns the reaped task ids.
|
||||
public async Task<IReadOnlyList<string>> ReapIdleAsync(DateTime nowUtc, TimeSpan idleTimeout)
|
||||
{
|
||||
var cutoffTicks = (nowUtc - idleTimeout).Ticks;
|
||||
List<string>? reaped = null;
|
||||
|
||||
foreach (var kvp in _sessions)
|
||||
{
|
||||
var entry = kvp.Value;
|
||||
if (entry.Session.IsTurnInFlight) continue;
|
||||
if (Interlocked.Read(ref entry.LastActivityTicksUtc) > cutoffTicks) continue;
|
||||
|
||||
if (_sessions.TryRemove(kvp.Key, out var removed))
|
||||
{
|
||||
try { await removed.Session.StopAsync(); }
|
||||
catch { /* already dead — leave it removed */ }
|
||||
(reaped ??= new()).Add(kvp.Key);
|
||||
}
|
||||
}
|
||||
|
||||
return reaped ?? (IReadOnlyList<string>)Array.Empty<string>();
|
||||
}
|
||||
}
|
||||
@@ -1,111 +0,0 @@
|
||||
using System.Diagnostics;
|
||||
using System.Text;
|
||||
using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
namespace ClaudeDo.Worker.Runner;
|
||||
|
||||
public sealed class ProcessClaudeStreamTransport : IClaudeStreamTransport
|
||||
{
|
||||
private readonly WorkerConfig _cfg;
|
||||
private readonly ILogger<ProcessClaudeStreamTransport> _logger;
|
||||
|
||||
private Process? _process;
|
||||
private Task? _stdoutTask;
|
||||
private Task? _stderrTask;
|
||||
|
||||
public event Func<string, Task>? LineReceived;
|
||||
public event Func<string, Task>? StderrReceived;
|
||||
|
||||
public ProcessClaudeStreamTransport(WorkerConfig cfg, ILogger<ProcessClaudeStreamTransport> logger)
|
||||
{
|
||||
_cfg = cfg;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public Task StartAsync(IReadOnlyList<string> args, string workingDirectory, CancellationToken ct)
|
||||
{
|
||||
var psi = new ProcessStartInfo
|
||||
{
|
||||
FileName = _cfg.ClaudeBin,
|
||||
WorkingDirectory = workingDirectory,
|
||||
RedirectStandardInput = true,
|
||||
RedirectStandardOutput = true,
|
||||
RedirectStandardError = true,
|
||||
UseShellExecute = false,
|
||||
CreateNoWindow = true,
|
||||
StandardOutputEncoding = Encoding.UTF8,
|
||||
StandardErrorEncoding = Encoding.UTF8,
|
||||
};
|
||||
|
||||
foreach (var arg in args)
|
||||
psi.ArgumentList.Add(arg);
|
||||
|
||||
psi.Environment["MCP_TOOL_TIMEOUT"] = "200000";
|
||||
|
||||
_process = new Process { StartInfo = psi };
|
||||
_process.Start();
|
||||
ProcessJobObject.Assign(_process, _logger);
|
||||
|
||||
// Keep stdin open — turns are driven by WriteLineAsync calls.
|
||||
_process.StandardInput.AutoFlush = false;
|
||||
|
||||
_stdoutTask = Task.Run(async () =>
|
||||
{
|
||||
while (await _process.StandardOutput.ReadLineAsync() is { } line)
|
||||
{
|
||||
if (string.IsNullOrEmpty(line)) continue;
|
||||
var handler = LineReceived;
|
||||
if (handler is not null)
|
||||
{
|
||||
try { await handler(line); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "LineReceived handler threw"); }
|
||||
}
|
||||
}
|
||||
}, CancellationToken.None);
|
||||
|
||||
_stderrTask = Task.Run(async () =>
|
||||
{
|
||||
while (await _process.StandardError.ReadLineAsync() is { } line)
|
||||
{
|
||||
if (string.IsNullOrEmpty(line)) continue;
|
||||
var handler = StderrReceived;
|
||||
if (handler is not null)
|
||||
{
|
||||
try { await handler(line); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "StderrReceived handler threw"); }
|
||||
}
|
||||
}
|
||||
}, CancellationToken.None);
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task WriteLineAsync(string jsonLine, CancellationToken ct)
|
||||
{
|
||||
if (_process is null) throw new InvalidOperationException("Transport not started.");
|
||||
await _process.StandardInput.WriteAsync((jsonLine + "\n").AsMemory(), ct);
|
||||
await _process.StandardInput.FlushAsync(ct);
|
||||
}
|
||||
|
||||
public void Kill()
|
||||
{
|
||||
try { _process?.Kill(entireProcessTree: true); }
|
||||
catch { /* already exited */ }
|
||||
}
|
||||
|
||||
public async Task WaitForExitAsync()
|
||||
{
|
||||
if (_process is not null)
|
||||
await _process.WaitForExitAsync(CancellationToken.None);
|
||||
if (_stdoutTask is not null) await _stdoutTask;
|
||||
if (_stderrTask is not null) await _stderrTask;
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
Kill();
|
||||
await WaitForExitAsync();
|
||||
_process?.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -1,207 +0,0 @@
|
||||
using System.Text.Json;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
namespace ClaudeDo.Worker.Runner;
|
||||
|
||||
public sealed class StreamingClaudeSession : ILiveSession
|
||||
{
|
||||
private readonly IClaudeStreamTransport _transport;
|
||||
private readonly Func<string, Task> _onLine;
|
||||
private readonly ILogger<StreamingClaudeSession> _logger;
|
||||
private readonly Action<IReadOnlyList<string>>? _onQueueChanged;
|
||||
private readonly Action<string>? _onUserMessageSent;
|
||||
|
||||
private readonly SemaphoreSlim _sendLock = new(1, 1);
|
||||
private volatile bool _isTurnInFlight;
|
||||
private readonly Queue<string> _pending = new();
|
||||
|
||||
public bool IsTurnInFlight => _isTurnInFlight;
|
||||
|
||||
public StreamingClaudeSession(
|
||||
IClaudeStreamTransport transport,
|
||||
Func<string, Task> onLine,
|
||||
ILogger<StreamingClaudeSession> logger,
|
||||
Action<IReadOnlyList<string>>? onQueueChanged = null,
|
||||
Action<string>? onUserMessageSent = null)
|
||||
{
|
||||
_transport = transport;
|
||||
_onLine = onLine;
|
||||
_logger = logger;
|
||||
_onQueueChanged = onQueueChanged;
|
||||
_onUserMessageSent = onUserMessageSent;
|
||||
}
|
||||
|
||||
private IReadOnlyList<string> SnapshotPending() => _pending.ToArray();
|
||||
|
||||
public async Task StartAsync(
|
||||
IReadOnlyList<string> args,
|
||||
string workingDirectory,
|
||||
string firstPrompt,
|
||||
CancellationToken ct)
|
||||
{
|
||||
_transport.LineReceived += HandleLineAsync;
|
||||
await _transport.StartAsync(args, workingDirectory, ct);
|
||||
await SendTurnAsync(firstPrompt, ct);
|
||||
_onUserMessageSent?.Invoke(firstPrompt);
|
||||
}
|
||||
|
||||
private async Task HandleLineAsync(string line)
|
||||
{
|
||||
try { await _onLine(line); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "onLine callback threw"); }
|
||||
|
||||
bool isResult;
|
||||
try
|
||||
{
|
||||
using var doc = JsonDocument.Parse(line);
|
||||
isResult = doc.RootElement.TryGetProperty("type", out var typeProp)
|
||||
&& typeProp.GetString() == "result";
|
||||
}
|
||||
catch { isResult = false; }
|
||||
|
||||
if (!isResult) return;
|
||||
|
||||
// Turn ended — flush one queued message if available.
|
||||
string? flushedText = null;
|
||||
IReadOnlyList<string>? remainingSnapshot = null;
|
||||
|
||||
await _sendLock.WaitAsync();
|
||||
try
|
||||
{
|
||||
_isTurnInFlight = false;
|
||||
if (_pending.Count > 0)
|
||||
{
|
||||
flushedText = _pending.Dequeue();
|
||||
remainingSnapshot = SnapshotPending();
|
||||
await SendTurnAsync(flushedText, CancellationToken.None);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_sendLock.Release();
|
||||
}
|
||||
|
||||
if (flushedText is not null)
|
||||
{
|
||||
_onQueueChanged?.Invoke(remainingSnapshot!);
|
||||
_onUserMessageSent?.Invoke(flushedText);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task SendUserMessageAsync(string text, CancellationToken ct)
|
||||
{
|
||||
bool enqueued = false;
|
||||
IReadOnlyList<string>? snapshot = null;
|
||||
|
||||
await _sendLock.WaitAsync(ct);
|
||||
try
|
||||
{
|
||||
if (_isTurnInFlight || _pending.Count > 0)
|
||||
{
|
||||
_pending.Enqueue(text);
|
||||
snapshot = SnapshotPending();
|
||||
enqueued = true;
|
||||
}
|
||||
else
|
||||
{
|
||||
await SendTurnAsync(text, ct);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_sendLock.Release();
|
||||
}
|
||||
|
||||
if (enqueued)
|
||||
_onQueueChanged?.Invoke(snapshot!);
|
||||
else
|
||||
_onUserMessageSent?.Invoke(text);
|
||||
}
|
||||
|
||||
public async Task RemoveQueuedAsync(string text, CancellationToken ct)
|
||||
{
|
||||
IReadOnlyList<string>? snapshot = null;
|
||||
|
||||
await _sendLock.WaitAsync(ct);
|
||||
try
|
||||
{
|
||||
if (_pending.Count == 0) return;
|
||||
|
||||
var list = _pending.ToList();
|
||||
var idx = list.IndexOf(text);
|
||||
if (idx < 0) return;
|
||||
|
||||
list.RemoveAt(idx);
|
||||
_pending.Clear();
|
||||
foreach (var item in list)
|
||||
_pending.Enqueue(item);
|
||||
|
||||
snapshot = SnapshotPending();
|
||||
}
|
||||
finally
|
||||
{
|
||||
_sendLock.Release();
|
||||
}
|
||||
|
||||
if (snapshot is not null)
|
||||
_onQueueChanged?.Invoke(snapshot);
|
||||
}
|
||||
|
||||
public async Task InterruptAsync(CancellationToken ct)
|
||||
{
|
||||
await _sendLock.WaitAsync(ct);
|
||||
try
|
||||
{
|
||||
if (!_isTurnInFlight) return;
|
||||
|
||||
var requestId = Guid.NewGuid().ToString();
|
||||
var payload = JsonSerializer.Serialize(new
|
||||
{
|
||||
type = "control_request",
|
||||
request_id = requestId,
|
||||
request = new { subtype = "interrupt" }
|
||||
});
|
||||
|
||||
try { await _transport.WriteLineAsync(payload, ct); }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "Failed to write interrupt control_request; degrading gracefully."); }
|
||||
}
|
||||
finally
|
||||
{
|
||||
_sendLock.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SendTurnAsync(string text, CancellationToken ct)
|
||||
{
|
||||
_isTurnInFlight = true;
|
||||
|
||||
var payload = JsonSerializer.Serialize(new
|
||||
{
|
||||
type = "user",
|
||||
message = new
|
||||
{
|
||||
role = "user",
|
||||
content = new[]
|
||||
{
|
||||
new { type = "text", text }
|
||||
}
|
||||
},
|
||||
parent_tool_use_id = (string?)null
|
||||
});
|
||||
|
||||
await _transport.WriteLineAsync(payload, ct);
|
||||
}
|
||||
|
||||
public async Task StopAsync()
|
||||
{
|
||||
_transport.Kill();
|
||||
await _transport.WaitForExitAsync();
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await StopAsync();
|
||||
await _transport.DisposeAsync();
|
||||
_sendLock.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -25,10 +25,6 @@ public abstract class StubWorkerClient : IWorkerClient
|
||||
public event Action<WorkerLogEntry>? WorkerLogReceivedEvent;
|
||||
public event Action<string, string, string>? TaskQuestionAskedEvent;
|
||||
public event Action<string, string>? TaskQuestionResolvedEvent;
|
||||
public event Action<string>? InteractiveSessionStartedEvent;
|
||||
public event Action<string>? InteractiveSessionEndedEvent;
|
||||
public event Action<string, IReadOnlyList<string>>? InteractiveQueueChangedEvent;
|
||||
public event Action<string, string>? InteractiveMessageSentEvent;
|
||||
public event Action? PrepStartedEvent;
|
||||
public event Action<string>? PrepLineEvent;
|
||||
public event Action<bool>? PrepFinishedEvent;
|
||||
@@ -59,11 +55,6 @@ public abstract class StubWorkerClient : IWorkerClient
|
||||
public void RaisePrepLine(string line) => PrepLineEvent?.Invoke(line);
|
||||
public void RaisePrepFinished(bool ok) => PrepFinishedEvent?.Invoke(ok);
|
||||
|
||||
public void RaiseInteractiveStarted(string taskId) => InteractiveSessionStartedEvent?.Invoke(taskId);
|
||||
public void RaiseInteractiveEnded(string taskId) => InteractiveSessionEndedEvent?.Invoke(taskId);
|
||||
public void RaiseInteractiveQueueChanged(string taskId, IReadOnlyList<string> pending) => InteractiveQueueChangedEvent?.Invoke(taskId, pending);
|
||||
public void RaiseInteractiveMessageSent(string taskId, string text) => InteractiveMessageSentEvent?.Invoke(taskId, text);
|
||||
|
||||
public virtual bool IsConnected => false;
|
||||
public virtual bool IsReconnecting => false;
|
||||
public virtual string? LastApproveTarget => null;
|
||||
@@ -103,7 +94,6 @@ public abstract class StubWorkerClient : IWorkerClient
|
||||
public virtual Task<MergeResultDto> ContinueConflictMergeAsync(string taskId) => Task.FromResult(new MergeResultDto("merged", System.Array.Empty<string>(), null));
|
||||
public virtual Task AbortConflictMergeAsync(string taskId) => Task.CompletedTask;
|
||||
public virtual Task StartPlanningSessionAsync(string taskId, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public virtual Task OpenInteractiveTerminalAsync(string taskId, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public virtual Task ResumeTaskInTerminalAsync(string taskId, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public virtual Task<LaunchSpec> GetInteractiveLaunchSpecAsync(string taskId, CancellationToken ct = default)
|
||||
=> Task.FromResult(new LaunchSpec(".", "claude", Array.Empty<string>(), new Dictionary<string, string>()));
|
||||
@@ -151,30 +141,6 @@ public abstract class StubWorkerClient : IWorkerClient
|
||||
public virtual Task SetOnlineInboxConfigAsync(OnlineInboxConfigInputDto input) => Task.CompletedTask;
|
||||
public virtual Task SetOnlineInboxAuthAsync(string refreshToken) => Task.CompletedTask;
|
||||
public virtual Task ClearOnlineInboxAuthAsync() => Task.CompletedTask;
|
||||
public List<(string TaskId, string Text)> SentInteractive { get; } = new();
|
||||
public virtual Task SendInteractiveMessageAsync(string taskId, string text)
|
||||
{
|
||||
SentInteractive.Add((taskId, text));
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
public List<(string TaskId, string Text)> RemovedQueued { get; } = new();
|
||||
public virtual Task RemoveQueuedInteractiveMessageAsync(string taskId, string text)
|
||||
{
|
||||
RemovedQueued.Add((taskId, text));
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
public List<string> StoppedInteractive { get; } = new();
|
||||
public virtual Task StopInteractiveSessionAsync(string taskId)
|
||||
{
|
||||
StoppedInteractive.Add(taskId);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
public List<string> InterruptedInteractive { get; } = new();
|
||||
public virtual Task InterruptInteractiveSessionAsync(string taskId)
|
||||
{
|
||||
InterruptedInteractive.Add(taskId);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
protected void RaisePropertyChanged(string name) => PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(name));
|
||||
}
|
||||
|
||||
@@ -189,224 +189,4 @@ public class TaskMonitorViewModelTests : IDisposable
|
||||
|
||||
Assert.False(vm.HasPendingQuestion);
|
||||
}
|
||||
|
||||
// ── Interactive composer ──────────────────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
public void InteractiveStarted_ForSubscribedTask_SetsIsInteractiveLive()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
|
||||
worker.RaiseInteractiveStarted("t1");
|
||||
|
||||
Assert.True(vm.IsInteractiveLive);
|
||||
Assert.Equal("running", vm.AgentState);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void InteractiveStarted_ForOtherTask_IsIgnored()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
|
||||
worker.RaiseInteractiveStarted("other");
|
||||
|
||||
Assert.False(vm.IsInteractiveLive);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void InteractiveEnded_ForSubscribedTask_ClearsIsInteractiveLive()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
worker.RaiseInteractiveStarted("t1");
|
||||
|
||||
worker.RaiseInteractiveEnded("t1");
|
||||
|
||||
Assert.False(vm.IsInteractiveLive);
|
||||
Assert.Equal("done", vm.AgentState);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void InteractiveEnded_ForOtherTask_IsIgnored()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
worker.RaiseInteractiveStarted("t1");
|
||||
|
||||
worker.RaiseInteractiveEnded("other");
|
||||
|
||||
Assert.True(vm.IsInteractiveLive); // unchanged
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SubmitComposerCommand_CanExecute_FalseWhenNotLive()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
vm.ComposerDraft = "hello";
|
||||
|
||||
Assert.False(vm.SubmitComposerCommand.CanExecute(null));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SubmitComposerCommand_CanExecute_FalseWhenLiveButDraftWhitespace()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
worker.RaiseInteractiveStarted("t1");
|
||||
vm.ComposerDraft = " ";
|
||||
|
||||
Assert.False(vm.SubmitComposerCommand.CanExecute(null));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SubmitComposerCommand_CanExecute_TrueWhenLiveAndDraftSet()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
worker.RaiseInteractiveStarted("t1");
|
||||
vm.ComposerDraft = "hello";
|
||||
|
||||
Assert.True(vm.SubmitComposerCommand.CanExecute(null));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SubmitComposer_CallsClient_ClearsDraft_DoesNotAddLogLine()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
worker.RaiseInteractiveStarted("t1");
|
||||
vm.ComposerDraft = "do the thing";
|
||||
|
||||
await vm.SubmitComposerCommand.ExecuteAsync(null);
|
||||
|
||||
Assert.Single(worker.SentInteractive);
|
||||
Assert.Equal(("t1", "do the thing"), worker.SentInteractive[0]);
|
||||
Assert.Equal(string.Empty, vm.ComposerDraft);
|
||||
// Log must NOT be updated by submit itself; it updates on InteractiveMessageSent
|
||||
Assert.Empty(vm.Log);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void InteractiveMessageSent_ForSubscribedTask_AddsUserLogLine()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
|
||||
worker.RaiseInteractiveMessageSent("t1", "hello from event");
|
||||
|
||||
Assert.Single(vm.Log);
|
||||
Assert.Equal(LogKind.User, vm.Log[0].Kind);
|
||||
Assert.Equal("hello from event", vm.Log[0].Text);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void InteractiveMessageSent_ForOtherTask_IsIgnored()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
|
||||
worker.RaiseInteractiveMessageSent("other", "not mine");
|
||||
|
||||
Assert.Empty(vm.Log);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void InteractiveQueueChanged_ForSubscribedTask_PopulatesQueue()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
|
||||
worker.RaiseInteractiveQueueChanged("t1", new[] { "msg1", "msg2" });
|
||||
|
||||
Assert.Equal(2, vm.QueuedMessages.Count);
|
||||
Assert.Equal("msg1", vm.QueuedMessages[0].Text);
|
||||
Assert.Equal("msg2", vm.QueuedMessages[1].Text);
|
||||
Assert.True(vm.HasQueuedMessages);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void InteractiveQueueChanged_EmptyList_ClearsQueue()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
worker.RaiseInteractiveQueueChanged("t1", new[] { "msg1" });
|
||||
|
||||
worker.RaiseInteractiveQueueChanged("t1", Array.Empty<string>());
|
||||
|
||||
Assert.Empty(vm.QueuedMessages);
|
||||
Assert.False(vm.HasQueuedMessages);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void InteractiveQueueChanged_ForOtherTask_IsIgnored()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
|
||||
worker.RaiseInteractiveQueueChanged("other", new[] { "msg1" });
|
||||
|
||||
Assert.Empty(vm.QueuedMessages);
|
||||
Assert.False(vm.HasQueuedMessages);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void InteractiveEnded_ClearsQueuedMessages()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
worker.RaiseInteractiveStarted("t1");
|
||||
worker.RaiseInteractiveQueueChanged("t1", new[] { "pending msg" });
|
||||
|
||||
worker.RaiseInteractiveEnded("t1");
|
||||
|
||||
Assert.Empty(vm.QueuedMessages);
|
||||
Assert.False(vm.HasQueuedMessages);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task QueuedMessageViewModel_RemoveCommand_RecordsRemoveCall()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
worker.RaiseInteractiveQueueChanged("t1", new[] { "a", "b" });
|
||||
|
||||
vm.QueuedMessages[0].RemoveCommand.Execute(null);
|
||||
// RemoveQueuedAsync is fire-and-forget; yield to let the async continuation run
|
||||
await System.Threading.Tasks.Task.Yield();
|
||||
|
||||
Assert.Single(worker.RemovedQueued);
|
||||
Assert.Equal(("t1", "a"), worker.RemovedQueued[0]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task InterruptInteractiveCommand_WhenLive_RecordsOneCall()
|
||||
{
|
||||
var worker = new FakeWorker();
|
||||
using var vm = Build(worker);
|
||||
vm.SetTaskId("t1");
|
||||
worker.RaiseInteractiveStarted("t1");
|
||||
|
||||
await vm.InterruptInteractiveCommand.ExecuteAsync(null);
|
||||
|
||||
Assert.Single(worker.InterruptedInteractive);
|
||||
Assert.Equal("t1", worker.InterruptedInteractive[0]);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,7 +21,7 @@ public sealed class ClearMyDayHubTests : IDisposable
|
||||
null!, null!, null!, null!, broadcaster, _db.CreateFactory(),
|
||||
null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!,
|
||||
null!, new ClaudeDo.Worker.Online.OnlineInboxConfig(), new ClaudeDo.Worker.Online.OnlineTokenStore(),
|
||||
new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), null!, null!);
|
||||
new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), null!);
|
||||
hub.Clients = new FakeHubCallerClients(new RecordingClientProxy());
|
||||
hub.Context = new FakeHubCallerContext();
|
||||
return hub;
|
||||
|
||||
@@ -31,7 +31,7 @@ public sealed class OnlineInboxHubTests : IDisposable
|
||||
var hub = new WorkerHub(
|
||||
null!, null!, null!, null!, broadcaster, null!,
|
||||
null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!,
|
||||
cfg, inboxCfg, store, new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), null!, null!);
|
||||
cfg, inboxCfg, store, new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), null!);
|
||||
hub.Clients = new FakeHubCallerClients(new RecordingClientProxy());
|
||||
hub.Context = new FakeHubCallerContext();
|
||||
return (hub, inboxCfg, store);
|
||||
|
||||
@@ -57,7 +57,7 @@ public sealed class PlanningHubTests : IDisposable
|
||||
null!, null!, null!, null!, null!, _db.CreateFactory(), null!, null!, null!,
|
||||
_planning, _launcher, null!, null!, null!, null!, null!, null!, null!, null!,
|
||||
null!, new ClaudeDo.Worker.Online.OnlineInboxConfig(), new ClaudeDo.Worker.Online.OnlineTokenStore(),
|
||||
new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), null!, null!);
|
||||
new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), null!);
|
||||
hub.Clients = new FakeHubCallerClients(_proxy);
|
||||
hub.Context = new FakeHubCallerContext();
|
||||
return hub;
|
||||
|
||||
@@ -52,7 +52,7 @@ public sealed class SessionSkillsHubTests : IDisposable
|
||||
null!, null!, null!, null!, broadcaster, _db.CreateFactory(),
|
||||
null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!,
|
||||
null!, new ClaudeDo.Worker.Online.OnlineInboxConfig(), new ClaudeDo.Worker.Online.OnlineTokenStore(),
|
||||
new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), null!, registry);
|
||||
new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), registry);
|
||||
hub.Clients = new FakeHubCallerClients(new RecordingClientProxy());
|
||||
hub.Context = new FakeHubCallerContext();
|
||||
return (hub, registry);
|
||||
|
||||
@@ -21,7 +21,7 @@ public sealed class WorktreeStateHubTests : IDisposable
|
||||
null!, null!, null!, null!, broadcaster, _db.CreateFactory(),
|
||||
null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!, null!,
|
||||
null!, new ClaudeDo.Worker.Online.OnlineInboxConfig(), new ClaudeDo.Worker.Online.OnlineTokenStore(),
|
||||
new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), null!, null!);
|
||||
new ClaudeDo.Worker.Runner.PendingQuestionRegistry(), null!);
|
||||
hub.Clients = new FakeHubCallerClients(new RecordingClientProxy());
|
||||
hub.Context = new FakeHubCallerContext();
|
||||
return hub;
|
||||
|
||||
@@ -1,47 +0,0 @@
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
namespace ClaudeDo.Worker.Tests.Infrastructure;
|
||||
|
||||
public sealed class FakeClaudeStreamTransport : IClaudeStreamTransport
|
||||
{
|
||||
public List<string> Written { get; } = [];
|
||||
public bool Killed { get; private set; }
|
||||
public bool Started { get; private set; }
|
||||
|
||||
public event Func<string, Task>? LineReceived;
|
||||
public event Func<string, Task>? StderrReceived;
|
||||
|
||||
public Task StartAsync(IReadOnlyList<string> args, string workingDirectory, CancellationToken ct)
|
||||
{
|
||||
Started = true;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task WriteLineAsync(string jsonLine, CancellationToken ct)
|
||||
{
|
||||
Written.Add(jsonLine);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public void Kill() => Killed = true;
|
||||
|
||||
public Task WaitForExitAsync() => Task.CompletedTask;
|
||||
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
|
||||
// Test helper: push a simulated stdout line to all LineReceived subscribers.
|
||||
public async Task PushLineAsync(string line)
|
||||
{
|
||||
var handler = LineReceived;
|
||||
if (handler is not null)
|
||||
await handler(line);
|
||||
}
|
||||
|
||||
// Test helper: push a simulated stderr line.
|
||||
public async Task PushStderrAsync(string line)
|
||||
{
|
||||
var handler = StderrReceived;
|
||||
if (handler is not null)
|
||||
await handler(line);
|
||||
}
|
||||
}
|
||||
@@ -1,303 +0,0 @@
|
||||
using ClaudeDo.Data;
|
||||
using ClaudeDo.Data.Models;
|
||||
using ClaudeDo.Data.Repositories;
|
||||
using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Hub;
|
||||
using ClaudeDo.Worker.Planning;
|
||||
using ClaudeDo.Worker.Runner;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
using ClaudeDo.Worker.Tests.Infrastructure;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||
|
||||
namespace ClaudeDo.Worker.Tests.Planning;
|
||||
|
||||
public sealed class InteractiveSessionServiceTests : IDisposable
|
||||
{
|
||||
private readonly DbFixture _db = new();
|
||||
private readonly ClaudeDoDbContext _ctx;
|
||||
private readonly TaskRepository _tasks;
|
||||
private readonly ListRepository _lists;
|
||||
private readonly CapturingHubContext _hubCtx;
|
||||
private readonly HubBroadcaster _broadcaster;
|
||||
private readonly LiveSessionRegistry _registry;
|
||||
private readonly WorkerConfig _cfg;
|
||||
|
||||
public InteractiveSessionServiceTests()
|
||||
{
|
||||
_ctx = _db.CreateContext();
|
||||
_tasks = new TaskRepository(_ctx);
|
||||
_lists = new ListRepository(_ctx);
|
||||
_hubCtx = new CapturingHubContext();
|
||||
_broadcaster = new HubBroadcaster(_hubCtx);
|
||||
_registry = new LiveSessionRegistry();
|
||||
_cfg = new WorkerConfig();
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
_ctx.Dispose();
|
||||
_db.Dispose();
|
||||
}
|
||||
|
||||
private InteractiveSessionService CreateService(
|
||||
Func<string, IReadOnlyList<string>, Func<string, Task>, (ILiveSession session, Task exitTask)>? factory = null)
|
||||
{
|
||||
return new InteractiveSessionService(
|
||||
_db.CreateFactory(),
|
||||
_cfg,
|
||||
_broadcaster,
|
||||
_registry,
|
||||
NullLoggerFactory.Instance,
|
||||
factory);
|
||||
}
|
||||
|
||||
private async Task<(string listId, string taskId, string workingDir)> SeedAsync()
|
||||
{
|
||||
var wd = Path.Combine(Path.GetTempPath(), $"iss_wd_{Guid.NewGuid():N}");
|
||||
Directory.CreateDirectory(wd);
|
||||
var listId = Guid.NewGuid().ToString();
|
||||
await _lists.AddAsync(new ListEntity
|
||||
{
|
||||
Id = listId,
|
||||
Name = "L",
|
||||
WorkingDir = wd,
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
});
|
||||
var task = new TaskEntity
|
||||
{
|
||||
Id = Guid.NewGuid().ToString(),
|
||||
ListId = listId,
|
||||
Title = "My task",
|
||||
Description = "Do the thing",
|
||||
Status = TaskStatus.Idle,
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
CommitType = "feat",
|
||||
};
|
||||
await _tasks.AddAsync(task);
|
||||
return (listId, task.Id, wd);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task StartAsync_MissingWorkingDir_Throws()
|
||||
{
|
||||
var listId = Guid.NewGuid().ToString();
|
||||
await _lists.AddAsync(new ListEntity
|
||||
{
|
||||
Id = listId,
|
||||
Name = "NoDir",
|
||||
WorkingDir = "/no/such/dir/ever/exists",
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
});
|
||||
var task = new TaskEntity
|
||||
{
|
||||
Id = Guid.NewGuid().ToString(),
|
||||
ListId = listId,
|
||||
Title = "T",
|
||||
Status = TaskStatus.Idle,
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
CommitType = "feat",
|
||||
};
|
||||
await _tasks.AddAsync(task);
|
||||
|
||||
var svc = CreateService();
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(
|
||||
() => svc.StartAsync(task.Id, CancellationToken.None));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task StartAsync_NullWorkingDir_Throws()
|
||||
{
|
||||
var listId = Guid.NewGuid().ToString();
|
||||
await _lists.AddAsync(new ListEntity
|
||||
{
|
||||
Id = listId,
|
||||
Name = "NullDir",
|
||||
WorkingDir = null,
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
});
|
||||
var task = new TaskEntity
|
||||
{
|
||||
Id = Guid.NewGuid().ToString(),
|
||||
ListId = listId,
|
||||
Title = "T",
|
||||
Status = TaskStatus.Idle,
|
||||
CreatedAt = DateTime.UtcNow,
|
||||
CommitType = "feat",
|
||||
};
|
||||
await _tasks.AddAsync(task);
|
||||
|
||||
var svc = CreateService();
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(
|
||||
() => svc.StartAsync(task.Id, CancellationToken.None));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task StartAsync_RegistersSessionAndBroadcastsStarted()
|
||||
{
|
||||
var (_, taskId, _) = await SeedAsync();
|
||||
var fakeSession = new FakeLiveSession();
|
||||
var exitTcs = new TaskCompletionSource<bool>();
|
||||
|
||||
var svc = CreateService((_, __, ___) => (fakeSession, exitTcs.Task));
|
||||
await svc.StartAsync(taskId, CancellationToken.None);
|
||||
|
||||
// Session registered
|
||||
Assert.True(_registry.TryGet(taskId, out var registered));
|
||||
Assert.Same(fakeSession, registered);
|
||||
|
||||
// InteractiveSessionStarted broadcast
|
||||
Assert.Contains(_hubCtx.Proxy.Calls, c => c.Method == "InteractiveSessionStarted");
|
||||
|
||||
// Cleanup
|
||||
exitTcs.SetResult(true);
|
||||
await Task.Delay(50); // let watcher fire
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task StartAsync_AlreadyRunning_Throws()
|
||||
{
|
||||
var (_, taskId, _) = await SeedAsync();
|
||||
var fakeSession = new FakeLiveSession();
|
||||
var exitTcs = new TaskCompletionSource<bool>();
|
||||
|
||||
var svc = CreateService((_, __, ___) => (fakeSession, exitTcs.Task));
|
||||
await svc.StartAsync(taskId, CancellationToken.None);
|
||||
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(
|
||||
() => svc.StartAsync(taskId, CancellationToken.None));
|
||||
|
||||
exitTcs.SetResult(true);
|
||||
await Task.Delay(50);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ExitWatcher_UnregistersAndBroadcastsEnded()
|
||||
{
|
||||
var (_, taskId, _) = await SeedAsync();
|
||||
var fakeSession = new FakeLiveSession();
|
||||
var exitTcs = new TaskCompletionSource<bool>();
|
||||
|
||||
var svc = CreateService((_, __, ___) => (fakeSession, exitTcs.Task));
|
||||
await svc.StartAsync(taskId, CancellationToken.None);
|
||||
|
||||
// Process exits naturally
|
||||
exitTcs.SetResult(true);
|
||||
|
||||
// Give the watcher time to run
|
||||
var deadline = DateTime.UtcNow.AddSeconds(2);
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
if (_registry.TryGet(taskId, out _) == false) break;
|
||||
await Task.Delay(10);
|
||||
}
|
||||
|
||||
Assert.False(_registry.TryGet(taskId, out _));
|
||||
Assert.Contains(_hubCtx.Proxy.Calls, c => c.Method == "InteractiveSessionEnded");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SendAsync_RoutesToSession()
|
||||
{
|
||||
var (_, taskId, _) = await SeedAsync();
|
||||
var fakeSession = new FakeLiveSession();
|
||||
var exitTcs = new TaskCompletionSource<bool>();
|
||||
|
||||
var svc = CreateService((_, __, ___) => (fakeSession, exitTcs.Task));
|
||||
await svc.StartAsync(taskId, CancellationToken.None);
|
||||
|
||||
await svc.SendAsync(taskId, "hello", CancellationToken.None);
|
||||
|
||||
Assert.Equal(1, fakeSession.SendCalls);
|
||||
Assert.Equal("hello", fakeSession.LastSentText);
|
||||
|
||||
exitTcs.SetResult(true);
|
||||
await Task.Delay(50);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SendAsync_NoSession_Throws()
|
||||
{
|
||||
var svc = CreateService();
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(
|
||||
() => svc.SendAsync("nonexistent-task", "text", CancellationToken.None));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task StopAsync_UnregistersSessionAndStopsIt()
|
||||
{
|
||||
var (_, taskId, _) = await SeedAsync();
|
||||
var fakeSession = new FakeLiveSession();
|
||||
// Keep the exit task pending so the exit watcher doesn't race with StopAsync.
|
||||
var exitTcs = new TaskCompletionSource<bool>();
|
||||
|
||||
var svc = CreateService((_, __, ___) => (fakeSession, exitTcs.Task));
|
||||
await svc.StartAsync(taskId, CancellationToken.None);
|
||||
|
||||
// Stop before the process exits naturally.
|
||||
await svc.StopAsync(taskId, CancellationToken.None);
|
||||
|
||||
// Registry is cleared by StopAsync (which calls _registry.StopAsync -> session.StopAsync + TryRemove).
|
||||
Assert.False(_registry.TryGet(taskId, out _));
|
||||
Assert.True(fakeSession.Stopped);
|
||||
|
||||
// Let the watcher complete harmlessly.
|
||||
exitTcs.SetResult(true);
|
||||
await Task.Delay(50);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task OnLineCallback_BroadcastsTaskMessageWithPrefix()
|
||||
{
|
||||
var (_, taskId, _) = await SeedAsync();
|
||||
Func<string, Task>? capturedOnLine = null;
|
||||
var fakeSession = new FakeLiveSession();
|
||||
var exitTcs = new TaskCompletionSource<bool>();
|
||||
|
||||
var svc = CreateService((_, __, onLine) =>
|
||||
{
|
||||
capturedOnLine = onLine;
|
||||
return (fakeSession, exitTcs.Task);
|
||||
});
|
||||
|
||||
await svc.StartAsync(taskId, CancellationToken.None);
|
||||
|
||||
Assert.NotNull(capturedOnLine);
|
||||
await capturedOnLine!("some line");
|
||||
|
||||
Assert.Contains(_hubCtx.Proxy.Calls, c =>
|
||||
c.Method == "TaskMessage" &&
|
||||
c.Args.Length >= 2 &&
|
||||
c.Args[1] is string s && s.StartsWith("[stdout] "));
|
||||
|
||||
exitTcs.SetResult(true);
|
||||
await Task.Delay(50);
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class FakeLiveSession : ILiveSession
|
||||
{
|
||||
public bool IsTurnInFlight => false;
|
||||
public int SendCalls { get; private set; }
|
||||
public string? LastSentText { get; private set; }
|
||||
public bool Stopped { get; private set; }
|
||||
|
||||
public Task SendUserMessageAsync(string text, CancellationToken ct)
|
||||
{
|
||||
SendCalls++;
|
||||
LastSentText = text;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task RemoveQueuedAsync(string text, CancellationToken ct) => Task.CompletedTask;
|
||||
|
||||
public Task InterruptAsync(CancellationToken ct) => Task.CompletedTask;
|
||||
|
||||
public Task StopAsync()
|
||||
{
|
||||
Stopped = true;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
}
|
||||
@@ -1,145 +0,0 @@
|
||||
using ClaudeDo.Worker.Runner;
|
||||
using ClaudeDo.Worker.Runner.Interfaces;
|
||||
|
||||
namespace ClaudeDo.Worker.Tests.Runner;
|
||||
|
||||
public sealed class LiveSessionRegistryTests
|
||||
{
|
||||
private sealed class FakeLiveSession : ILiveSession
|
||||
{
|
||||
public bool StopCalled { get; private set; }
|
||||
public bool IsTurnInFlight { get; set; }
|
||||
|
||||
public Task SendUserMessageAsync(string text, CancellationToken ct) => Task.CompletedTask;
|
||||
public Task RemoveQueuedAsync(string text, CancellationToken ct) => Task.CompletedTask;
|
||||
public Task InterruptAsync(CancellationToken ct) => Task.CompletedTask;
|
||||
|
||||
public Task StopAsync()
|
||||
{
|
||||
StopCalled = true;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Register_ThenTryGet_ReturnsSession()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
var session = new FakeLiveSession();
|
||||
|
||||
registry.Register("task-1", session);
|
||||
|
||||
Assert.True(registry.TryGet("task-1", out var retrieved));
|
||||
Assert.Same(session, retrieved);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void TryGet_Missing_ReturnsFalse()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
|
||||
Assert.False(registry.TryGet("no-such-task", out _));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Unregister_RemovesSession()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
registry.Register("task-1", new FakeLiveSession());
|
||||
registry.Unregister("task-1");
|
||||
|
||||
Assert.False(registry.TryGet("task-1", out _));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Register_WhenSessionAlreadyExists_StopsPreviousSession()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
var first = new FakeLiveSession();
|
||||
var second = new FakeLiveSession();
|
||||
|
||||
registry.Register("task-1", first);
|
||||
registry.Register("task-1", second);
|
||||
|
||||
// Give the fire-and-forget stop a tick to run.
|
||||
await Task.Delay(50);
|
||||
|
||||
Assert.True(first.StopCalled);
|
||||
Assert.True(registry.TryGet("task-1", out var retrieved));
|
||||
Assert.Same(second, retrieved);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task StopAsync_StopsAndRemovesSession()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
var session = new FakeLiveSession();
|
||||
registry.Register("task-1", session);
|
||||
|
||||
await registry.StopAsync("task-1");
|
||||
|
||||
Assert.True(session.StopCalled);
|
||||
Assert.False(registry.TryGet("task-1", out _));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task StopAsync_MissingTask_DoesNotThrow()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
await registry.StopAsync("no-such-task"); // should not throw
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReapIdleAsync_StopsAndRemovesIdleSession()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
var session = new FakeLiveSession();
|
||||
registry.Register("task-1", session);
|
||||
|
||||
// Sweep "now" is an hour past registration, well beyond the 30-min idle window.
|
||||
var reaped = await registry.ReapIdleAsync(DateTime.UtcNow.AddMinutes(60), TimeSpan.FromMinutes(30));
|
||||
|
||||
Assert.Contains("task-1", reaped);
|
||||
Assert.True(session.StopCalled);
|
||||
Assert.False(registry.TryGet("task-1", out _));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReapIdleAsync_KeepsRecentlyActiveSession()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
var session = new FakeLiveSession();
|
||||
registry.Register("task-1", session);
|
||||
|
||||
var reaped = await registry.ReapIdleAsync(DateTime.UtcNow, TimeSpan.FromMinutes(30));
|
||||
|
||||
Assert.Empty(reaped);
|
||||
Assert.False(session.StopCalled);
|
||||
Assert.True(registry.TryGet("task-1", out _));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReapIdleAsync_SkipsSessionWithTurnInFlight()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
var session = new FakeLiveSession { IsTurnInFlight = true };
|
||||
registry.Register("task-1", session);
|
||||
|
||||
// Idle long enough to reap, but a turn is in flight → must be left alone.
|
||||
var reaped = await registry.ReapIdleAsync(DateTime.UtcNow.AddMinutes(60), TimeSpan.FromMinutes(30));
|
||||
|
||||
Assert.Empty(reaped);
|
||||
Assert.False(session.StopCalled);
|
||||
Assert.True(registry.TryGet("task-1", out _));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReapIdleAsync_NoSessions_ReturnsEmpty()
|
||||
{
|
||||
var registry = new LiveSessionRegistry();
|
||||
var reaped = await registry.ReapIdleAsync(DateTime.UtcNow, TimeSpan.FromMinutes(30));
|
||||
Assert.Empty(reaped);
|
||||
}
|
||||
}
|
||||
@@ -1,469 +0,0 @@
|
||||
using System.Text.Json;
|
||||
using ClaudeDo.Worker.Runner;
|
||||
using ClaudeDo.Worker.Tests.Infrastructure;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
|
||||
namespace ClaudeDo.Worker.Tests.Runner;
|
||||
|
||||
public sealed class StreamingClaudeSessionTests
|
||||
{
|
||||
private static StreamingClaudeSession Build(
|
||||
FakeClaudeStreamTransport transport,
|
||||
List<string> received)
|
||||
{
|
||||
return new StreamingClaudeSession(
|
||||
transport,
|
||||
line => { received.Add(line); return Task.CompletedTask; },
|
||||
NullLogger<StreamingClaudeSession>.Instance);
|
||||
}
|
||||
|
||||
private static string ResultLine(bool isError = false, string subtype = "success") =>
|
||||
JsonSerializer.Serialize(new { type = "result", is_error = isError, subtype });
|
||||
|
||||
private static string UserMessageLine(string text) =>
|
||||
JsonSerializer.Serialize(new
|
||||
{
|
||||
type = "user",
|
||||
message = new { role = "user", content = new[] { new { type = "text", text } } },
|
||||
parent_tool_use_id = (string?)null
|
||||
});
|
||||
|
||||
// ---- Start sends first prompt as user-message, IsTurnInFlight = true ----
|
||||
|
||||
[Fact]
|
||||
public async Task Start_SendsFirstPromptAsUserMessage_AndTurnIsInFlight()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var received = new List<string>();
|
||||
var session = Build(transport, received);
|
||||
|
||||
await session.StartAsync([], "/tmp", "hello world", CancellationToken.None);
|
||||
|
||||
Assert.True(session.IsTurnInFlight);
|
||||
Assert.Single(transport.Written);
|
||||
|
||||
using var doc = JsonDocument.Parse(transport.Written[0]);
|
||||
var root = doc.RootElement;
|
||||
Assert.Equal("user", root.GetProperty("type").GetString());
|
||||
var text = root.GetProperty("message").GetProperty("content")[0].GetProperty("text").GetString();
|
||||
Assert.Equal("hello world", text);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ---- Pushing a result line flips IsTurnInFlight to false ----
|
||||
|
||||
[Fact]
|
||||
public async Task PushingResultLine_FlipsIsTurnInFlightToFalse()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var session = Build(transport, []);
|
||||
|
||||
await session.StartAsync([], "/tmp", "prompt", CancellationToken.None);
|
||||
Assert.True(session.IsTurnInFlight);
|
||||
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
|
||||
Assert.False(session.IsTurnInFlight);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ---- Sending while in-flight queues the message; no interrupt written ----
|
||||
|
||||
[Fact]
|
||||
public async Task SendWhileInFlight_QueuesMessage_NoInterrupt()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var session = Build(transport, []);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
// Written[0] = first user message. Turn is in flight.
|
||||
Assert.True(session.IsTurnInFlight);
|
||||
var countBefore = transport.Written.Count;
|
||||
|
||||
await session.SendUserMessageAsync("second", CancellationToken.None);
|
||||
|
||||
// Nothing extra written yet — message is queued, no interrupt issued.
|
||||
Assert.Equal(countBefore, transport.Written.Count);
|
||||
Assert.True(session.IsTurnInFlight);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ---- Queued message flushes automatically when result arrives ----
|
||||
|
||||
[Fact]
|
||||
public async Task QueuedMessage_FlushesOnResult()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var session = Build(transport, []);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
await session.SendUserMessageAsync("second", CancellationToken.None);
|
||||
|
||||
// Push result — should dequeue "second" and send it.
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
|
||||
// After flush: IsTurnInFlight is true again for the second turn.
|
||||
Assert.True(session.IsTurnInFlight);
|
||||
|
||||
// Written[0] = "first", Written[1] = "second" user message.
|
||||
Assert.Equal(2, transport.Written.Count);
|
||||
using var doc = JsonDocument.Parse(transport.Written[1]);
|
||||
Assert.Equal("user", doc.RootElement.GetProperty("type").GetString());
|
||||
var text = doc.RootElement.GetProperty("message").GetProperty("content")[0].GetProperty("text").GetString();
|
||||
Assert.Equal("second", text);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ---- Interrupt writes control_request when in-flight ----
|
||||
|
||||
[Fact]
|
||||
public async Task Interrupt_WritesControlRequest_WhenInFlight()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var session = Build(transport, []);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
await session.SendUserMessageAsync("second", CancellationToken.None); // queued
|
||||
|
||||
await session.InterruptAsync(CancellationToken.None);
|
||||
|
||||
// Written[0] = first user message, Written[1] = interrupt control_request.
|
||||
Assert.True(transport.Written.Count >= 2);
|
||||
using var interruptDoc = JsonDocument.Parse(transport.Written[1]);
|
||||
Assert.Equal("control_request", interruptDoc.RootElement.GetProperty("type").GetString());
|
||||
Assert.Equal("interrupt", interruptDoc.RootElement.GetProperty("request").GetProperty("subtype").GetString());
|
||||
|
||||
// Now push result — queued "second" must flush.
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
|
||||
Assert.True(session.IsTurnInFlight);
|
||||
using var userDoc = JsonDocument.Parse(transport.Written[^1]);
|
||||
Assert.Equal("user", userDoc.RootElement.GetProperty("type").GetString());
|
||||
var text = userDoc.RootElement.GetProperty("message").GetProperty("content")[0].GetProperty("text").GetString();
|
||||
Assert.Equal("second", text);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ---- Interrupt is a no-op when idle ----
|
||||
|
||||
[Fact]
|
||||
public async Task Interrupt_NoOp_WhenIdle()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var session = Build(transport, []);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
await transport.PushLineAsync(ResultLine()); // idle now
|
||||
Assert.False(session.IsTurnInFlight);
|
||||
|
||||
var countBefore = transport.Written.Count;
|
||||
await session.InterruptAsync(CancellationToken.None);
|
||||
|
||||
Assert.Equal(countBefore, transport.Written.Count);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ---- Sending while idle writes user message with no interrupt ----
|
||||
|
||||
[Fact]
|
||||
public async Task SendWhileIdle_WritesUserMessageWithNoInterrupt()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var session = Build(transport, []);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
await transport.PushLineAsync(ResultLine()); // end the turn → idle
|
||||
|
||||
Assert.False(session.IsTurnInFlight);
|
||||
var countBefore = transport.Written.Count;
|
||||
|
||||
await session.SendUserMessageAsync("second", CancellationToken.None);
|
||||
|
||||
// Exactly one new write, no interrupt.
|
||||
Assert.Equal(countBefore + 1, transport.Written.Count);
|
||||
using var doc = JsonDocument.Parse(transport.Written[^1]);
|
||||
Assert.Equal("user", doc.RootElement.GetProperty("type").GetString());
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ---- Result with is_error / error_during_execution still ends the turn ----
|
||||
|
||||
[Fact]
|
||||
public async Task ResultWithIsError_StillEndsTurn_NoThrow()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var session = Build(transport, []);
|
||||
|
||||
await session.StartAsync([], "/tmp", "prompt", CancellationToken.None);
|
||||
Assert.True(session.IsTurnInFlight);
|
||||
|
||||
await transport.PushLineAsync(ResultLine(isError: true, subtype: "error_during_execution"));
|
||||
|
||||
Assert.False(session.IsTurnInFlight);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ---- onLine receives every pushed stdout line ----
|
||||
|
||||
[Fact]
|
||||
public async Task OnLine_ReceivesEveryPushedLine()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var received = new List<string>();
|
||||
var session = Build(transport, received);
|
||||
|
||||
await session.StartAsync([], "/tmp", "prompt", CancellationToken.None);
|
||||
|
||||
var lines = new[] { "{\"type\":\"assistant\"}", "{\"type\":\"stream_event\"}", ResultLine() };
|
||||
foreach (var l in lines)
|
||||
await transport.PushLineAsync(l);
|
||||
|
||||
Assert.Equal(lines, received);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ---- Multiple queued messages flush one-per-result in FIFO order ----
|
||||
|
||||
[Fact]
|
||||
public async Task MultipleQueued_FlushInFifoOrder()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var session = Build(transport, []);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
await session.SendUserMessageAsync("second", CancellationToken.None);
|
||||
await session.SendUserMessageAsync("third", CancellationToken.None);
|
||||
|
||||
// Both "second" and "third" are queued; nothing extra written yet.
|
||||
Assert.Single(transport.Written);
|
||||
|
||||
// Result 1 → flushes "second".
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
Assert.Equal(2, transport.Written.Count);
|
||||
using var doc2 = JsonDocument.Parse(transport.Written[1]);
|
||||
Assert.Equal("second", doc2.RootElement.GetProperty("message").GetProperty("content")[0].GetProperty("text").GetString());
|
||||
Assert.True(session.IsTurnInFlight);
|
||||
|
||||
// Result 2 → flushes "third".
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
Assert.Equal(3, transport.Written.Count);
|
||||
using var doc3 = JsonDocument.Parse(transport.Written[2]);
|
||||
Assert.Equal("third", doc3.RootElement.GetProperty("message").GetProperty("content")[0].GetProperty("text").GetString());
|
||||
Assert.True(session.IsTurnInFlight);
|
||||
|
||||
// Result 3 → queue empty, idle.
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
Assert.False(session.IsTurnInFlight);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────────────────────
|
||||
// Callback tests (onQueueChanged / onUserMessageSent)
|
||||
// ──────────────────────────────────────────────────────────────────────────
|
||||
|
||||
private static StreamingClaudeSession BuildWithCallbacks(
|
||||
FakeClaudeStreamTransport transport,
|
||||
List<IReadOnlyList<string>> queueChanges,
|
||||
List<string> sent)
|
||||
{
|
||||
return new StreamingClaudeSession(
|
||||
transport,
|
||||
line => Task.CompletedTask,
|
||||
NullLogger<StreamingClaudeSession>.Instance,
|
||||
onQueueChanged: snapshot => queueChanges.Add(snapshot),
|
||||
onUserMessageSent: text => sent.Add(text));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Start_InvokesOnUserMessageSent_WithFirstPrompt()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var sent = new List<string>();
|
||||
var session = BuildWithCallbacks(transport, [], sent);
|
||||
|
||||
await session.StartAsync([], "/tmp", "hello", CancellationToken.None);
|
||||
|
||||
Assert.Single(sent);
|
||||
Assert.Equal("hello", sent[0]);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SendWhileInFlight_InvokesOnQueueChanged_NotOnUserMessageSent()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var queueChanges = new List<IReadOnlyList<string>>();
|
||||
var sent = new List<string>();
|
||||
var session = BuildWithCallbacks(transport, queueChanges, sent);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
sent.Clear(); // ignore the initial prompt notification
|
||||
|
||||
await session.SendUserMessageAsync("queued-msg", CancellationToken.None);
|
||||
|
||||
Assert.Single(queueChanges);
|
||||
Assert.Contains("queued-msg", queueChanges[0]);
|
||||
Assert.DoesNotContain("queued-msg", sent);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PushResult_FlushesPending_InvokesQueueClearThenUserMessageSent()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var queueChanges = new List<IReadOnlyList<string>>();
|
||||
var sent = new List<string>();
|
||||
var session = BuildWithCallbacks(transport, queueChanges, sent);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
sent.Clear();
|
||||
|
||||
await session.SendUserMessageAsync("queued-msg", CancellationToken.None);
|
||||
queueChanges.Clear(); // ignore the enqueue snapshot
|
||||
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
|
||||
// After flush: one queueChanged with empty list, then sent contains flushed text.
|
||||
Assert.Single(queueChanges);
|
||||
Assert.Empty(queueChanges[0]);
|
||||
Assert.Single(sent);
|
||||
Assert.Equal("queued-msg", sent[0]);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SendWhileIdle_InvokesOnUserMessageSent_NoQueueChanged()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var queueChanges = new List<IReadOnlyList<string>>();
|
||||
var sent = new List<string>();
|
||||
var session = BuildWithCallbacks(transport, queueChanges, sent);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
await transport.PushLineAsync(ResultLine()); // go idle
|
||||
sent.Clear();
|
||||
queueChanges.Clear();
|
||||
|
||||
await session.SendUserMessageAsync("idle-msg", CancellationToken.None);
|
||||
|
||||
Assert.Empty(queueChanges);
|
||||
Assert.Single(sent);
|
||||
Assert.Equal("idle-msg", sent[0]);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task TwoMessagesQueued_FlushFifo_QueueSnapshotsShrink()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var queueChanges = new List<IReadOnlyList<string>>();
|
||||
var sent = new List<string>();
|
||||
var session = BuildWithCallbacks(transport, queueChanges, sent);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
sent.Clear();
|
||||
|
||||
await session.SendUserMessageAsync("second", CancellationToken.None);
|
||||
await session.SendUserMessageAsync("third", CancellationToken.None);
|
||||
// queueChanges[0] = ["second"], queueChanges[1] = ["second","third"]
|
||||
Assert.Equal(2, queueChanges.Count);
|
||||
Assert.Equal(new[] { "second" }, queueChanges[0]);
|
||||
Assert.Equal(new[] { "second", "third" }, queueChanges[1]);
|
||||
queueChanges.Clear();
|
||||
|
||||
// Result 1 → flushes "second"; remaining queue = ["third"]
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
Assert.Single(queueChanges);
|
||||
Assert.Equal(new[] { "third" }, queueChanges[0]);
|
||||
Assert.Single(sent);
|
||||
Assert.Equal("second", sent[0]);
|
||||
sent.Clear();
|
||||
queueChanges.Clear();
|
||||
|
||||
// Result 2 → flushes "third"; remaining queue = []
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
Assert.Single(queueChanges);
|
||||
Assert.Empty(queueChanges[0]);
|
||||
Assert.Single(sent);
|
||||
Assert.Equal("third", sent[0]);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────────────────────
|
||||
// RemoveQueuedAsync tests
|
||||
// ──────────────────────────────────────────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
public async Task RemoveQueued_RemovesFirstOccurrence_SnapshotContainsOnlySecond_AndSecondDeliveredOnResult()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var queueChanges = new List<IReadOnlyList<string>>();
|
||||
var sent = new List<string>();
|
||||
var session = BuildWithCallbacks(transport, queueChanges, sent);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
sent.Clear();
|
||||
|
||||
// Enqueue two messages while turn is in flight.
|
||||
await session.SendUserMessageAsync("alpha", CancellationToken.None);
|
||||
await session.SendUserMessageAsync("beta", CancellationToken.None);
|
||||
queueChanges.Clear();
|
||||
|
||||
// Remove "alpha" from the queue.
|
||||
await session.RemoveQueuedAsync("alpha", CancellationToken.None);
|
||||
|
||||
// Snapshot emitted and contains only "beta".
|
||||
Assert.Single(queueChanges);
|
||||
Assert.Equal(new[] { "beta" }, queueChanges[0]);
|
||||
|
||||
// Push result → only "beta" is flushed, not "alpha".
|
||||
sent.Clear();
|
||||
queueChanges.Clear();
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
|
||||
Assert.Single(sent);
|
||||
Assert.Equal("beta", sent[0]);
|
||||
|
||||
// Queue now empty; next result leaves us idle.
|
||||
await transport.PushLineAsync(ResultLine());
|
||||
Assert.False(session.IsTurnInFlight);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RemoveQueued_NotFound_NoQueueChangedCallback()
|
||||
{
|
||||
var transport = new FakeClaudeStreamTransport();
|
||||
var queueChanges = new List<IReadOnlyList<string>>();
|
||||
var sent = new List<string>();
|
||||
var session = BuildWithCallbacks(transport, queueChanges, sent);
|
||||
|
||||
await session.StartAsync([], "/tmp", "first", CancellationToken.None);
|
||||
await session.SendUserMessageAsync("alpha", CancellationToken.None);
|
||||
queueChanges.Clear();
|
||||
|
||||
// Try to remove a message that is not in the queue.
|
||||
await session.RemoveQueuedAsync("nope", CancellationToken.None);
|
||||
|
||||
// No new snapshot emitted.
|
||||
Assert.Empty(queueChanges);
|
||||
|
||||
await session.DisposeAsync();
|
||||
}
|
||||
}
|
||||
@@ -36,10 +36,6 @@ sealed class FakeWorkerClient : IWorkerClient
|
||||
public event Action<WorkerLogEntry>? WorkerLogReceivedEvent;
|
||||
public event Action<string, string, string>? TaskQuestionAskedEvent;
|
||||
public event Action<string, string>? TaskQuestionResolvedEvent;
|
||||
public event Action<string>? InteractiveSessionStartedEvent;
|
||||
public event Action<string>? InteractiveSessionEndedEvent;
|
||||
public event Action<string, IReadOnlyList<string>>? InteractiveQueueChangedEvent;
|
||||
public event Action<string, string>? InteractiveMessageSentEvent;
|
||||
public void RaiseTaskUpdated(string taskId) => TaskUpdatedEvent?.Invoke(taskId);
|
||||
public void RaiseWorktreeUpdated(string taskId) => WorktreeUpdatedEvent?.Invoke(taskId);
|
||||
public void RaiseTaskMessage(string taskId, string line) => TaskMessageEvent?.Invoke(taskId, line);
|
||||
@@ -73,7 +69,6 @@ sealed class FakeWorkerClient : IWorkerClient
|
||||
public Task CancelReviewAsync(string taskId) => Task.CompletedTask;
|
||||
public Task WakeQueueAsync() { WakeQueueCalls++; return Task.CompletedTask; }
|
||||
public Task StartPlanningSessionAsync(string taskId, CancellationToken ct = default) { StartPlanningCalls++; return Task.CompletedTask; }
|
||||
public Task OpenInteractiveTerminalAsync(string taskId, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public int PickUpInTerminalCalls { get; private set; }
|
||||
public string? LastPickUpTaskId { get; private set; }
|
||||
public Task ResumeTaskInTerminalAsync(string taskId, CancellationToken ct = default) { PickUpInTerminalCalls++; LastPickUpTaskId = taskId; return Task.CompletedTask; }
|
||||
@@ -137,10 +132,6 @@ sealed class FakeWorkerClient : IWorkerClient
|
||||
public Task SetOnlineInboxConfigAsync(OnlineInboxConfigInputDto input) => Task.CompletedTask;
|
||||
public Task SetOnlineInboxAuthAsync(string refreshToken) => Task.CompletedTask;
|
||||
public Task ClearOnlineInboxAuthAsync() => Task.CompletedTask;
|
||||
public Task SendInteractiveMessageAsync(string taskId, string text) => Task.CompletedTask;
|
||||
public Task RemoveQueuedInteractiveMessageAsync(string taskId, string text) => Task.CompletedTask;
|
||||
public Task StopInteractiveSessionAsync(string taskId) => Task.CompletedTask;
|
||||
public Task InterruptInteractiveSessionAsync(string taskId) => Task.CompletedTask;
|
||||
public IReadOnlyList<ActiveTask> GetActiveTasks() => System.Array.Empty<ActiveTask>();
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user