From 3062857db6c8b5027e9b98d5a910babc92fa96bc Mon Sep 17 00:00:00 2001 From: d3cker Date: Wed, 16 Sep 2026 07:58:48 +0200 Subject: [PATCH] feat: manage and durably close bot tasks from TUI --- AGENTS.md | 4 +- CHANGELOG.md | 7 +++ README.md | 4 ++ docs/advanced.md | 25 +++++++++ docs/architecture.md | 7 +++ docs/bot-workflow.md | 80 +++++++++++++++++++++++++---- docs/installation.md | 2 +- docs/runtime.md | 49 +++++++++++++++++- src/activity.ts | 5 +- src/dispatcher.ts | 101 +++++++++++++++++++++++++++++++------ src/executor.ts | 46 +++++++++++------ src/plugins/github.ts | 1 + src/rpc.ts | 1 + src/runtime-panel.ts | 10 ++-- src/ui.ts | 59 +++++++++++++++++++--- test/core.test.ts | 73 +++++++++++++++++++++++++++ test/executor.test.ts | 24 +++++++++ test/runtime-panel.test.ts | 11 ++++ test/ui.test.ts | 32 ++++++++++-- 19 files changed, 481 insertions(+), 60 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 69315cd..b3cd2e9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,7 +28,7 @@ steps, project setup, headless operation, and removal. | [docs/architecture.md](docs/architecture.md) | Component responsibilities, a short issue-to-PR overview, configuration ownership, scheduler ownership, and shared state. | Start here to understand how the system is divided before locating implementation code. | | [docs/bot-workflow.md](docs/bot-workflow.md) | Eight Mermaid diagrams and detailed implementation notes: startup and polling; discovery and routing; task phases; sessions and questions; media helpers; verification and publication; feedback, merging, and tab closure; status, retries, and recovery. Includes links to the source for each area. | Use for exact execution order, state transitions, checkpoint behavior, failure paths, and tracing a bot task from issue to merged PR. | | [docs/configuration.md](docs/configuration.md) | The standard `.opencode/automation.json` format, defaults, setup flags, configuration tracking across Git branches, authors, triggers, checks, base branches, model capabilities, media helpers, custom prompts, signatures, and auto-merge settings. | Use when adding or changing user-facing configuration, defaults, or setup examples. | -| [docs/runtime.md](docs/runtime.md) | User-visible behavior while the bot runs: GitHub questions and permission replies, branch selection, media inputs, prompt loading, follow-up comments, session tabs, runtime sidebar/status freshness, and routine management commands. | Use when changing issue conversations, session continuation, runtime tools, or TUI behavior. | +| [docs/runtime.md](docs/runtime.md) | User-visible behavior while the bot runs: GitHub questions and permission replies, branch selection, media inputs, prompt loading, follow-up comments, session tabs, runtime sidebar/status freshness, local task closure, and routine management commands. | Use when changing issue conversations, session continuation, runtime tools, or TUI behavior. | | [docs/advanced.md](docs/advanced.md) | Separate scheduler/dispatcher setup, multiple repositories, custom RPC jobs, full options, timeouts, management and retry commands, persistence, reconciliation, locks, and known limits. | Use for low-level configuration, operational troubleshooting, recovery, or ownership/concurrency changes. | | [docs/installation.md](docs/installation.md) | Loader registration, config-directory precedence, prerequisites, source installation, project-local installation, upgrade conflicts, testing on another machine, and migration limits. | Use when working on packaging, installers, registration, upgrades, or deployment troubleshooting. | | [docs/releases.md](docs/releases.md) | Feature-to-devel and devel-to-release PR checks, automatic patch versions, manual npm version/tag releases, exact changelog notes, publication recovery, README commits on release, automatic release-to-devel synchronization, and promotion PRs into protected main. | Use for CI triggers, versioning, packaging, GitHub Release publication, branch permissions, or recovery after a failed release. | @@ -58,7 +58,7 @@ the installation block without making remote writes. Keep its markers intact. and GitHub plugin entrypoints. `src/easy.ts` resolves standard project settings; `src/config.ts` defines the configuration schemas and route matching. - `src/dispatcher.ts` owns discovery, the durable task lifecycle, questions, - feedback rounds, publication coordination, retries, and merge polling. + feedback rounds, publication coordination, retries, durable task closure, and merge polling. `src/scheduler.ts` owns interval jobs; `src/state.ts` owns persistence and locks. - `src/executor.ts` owns analysis, base selection, worktrees, session execution, verification, and pushing. `src/analysis.ts` and `src/branch.ts` validate model diff --git a/CHANGELOG.md b/CHANGELOG.md index ab6379d..8a1fbe7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,13 @@ include the full version, for example `## 0.7.0-beta.1`. ### Added +- Manage tasks directly from `/bot`: inspect details, open sessions, close idle + tabs, restart workflows, or stop sessions and durably end tracking without + deleting work. Preserve closed tasks as history and skip rediscovery, feedback, + runtime hooks and publication after closure, including missing issue/PR cases. +- Identify blocked, failed and closing issue keys and errors in the runtime + sidebar instead of showing only an anonymous attention counter. + - Add a live BOT RUNTIME sidebar and `/botstatus` report with dispatcher operations, scheduler scans/retries, queue counts and selected-task details. Keep stale and unavailable readings explicit; monitor through read-only owner-scoped RPC. diff --git a/README.md b/README.md index 4d6bcf2..94d1399 100644 --- a/README.md +++ b/README.md @@ -254,6 +254,10 @@ installations are not removed by `npm uninstall --global`. `/botstatus` opens a full text report. Status refreshes every five seconds; unavailable or stale readings are marked explicitly. See [runtime panel details](docs/runtime.md#runtime-status-sidebar). +- **Task management:** `/bot` lets you open a session, inspect details, close idle + tabs, restart a stopped workflow, or stop sessions and end task tracking. Closing + tracking preserves all local work and history, works without a surviving GitHub + issue/PR, and prevents rediscovery. See [task management](docs/runtime.md#manage-tasks-from-bot). - **Progress:** use `/bot` in the TUI, or the CLI's `status`, `scan`, `pause`, and `resume` commands from the target repository. Closing a PR closes its bot tabs while retaining session history. Authorized issue comments can continue work diff --git a/docs/advanced.md b/docs/advanced.md index 66f3762..4c8941b 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -182,3 +182,28 @@ API references: [OpenCode 2 plugins](https://opencode.ai/v2/docs/build/plugins), [GitHub issues](https://docs.github.com/en/rest/issues/issues), [comments](https://docs.github.com/en/rest/issues/comments), and [pull requests](https://docs.github.com/en/rest/pulls/pulls). + +## Ending task tracking + +Use `/bot` → select issue → **Stop and close task**. The owner-scoped RPC is +`automation.github.close` with `{ "key": "owner/repository#123" }`, returning +`{ "accepted": true }` when durable closure is queued or `false` if already closed. +There is no corresponding setup CLI subcommand. This action does not require the +GitHub issue/PR or saved session to still exist. It never deletes local work. + +The queue retains phase and history with statuses `closing` and `closed`, +`closeRequestedAt`, `closedAt`, and `closeError`. Interruption of all saved main, +earlier-round and media session IDs is bounded to 15 seconds per request; missing +sessions are ignored, other failures retry no sooner than 30 seconds. Closure +waits for the selected task's in-flight worker and question posts, then interrupts +again to cover a session creation that was already in flight. Checkpoint guards +prevent late results from publishing or reviving the task. Publication/merge +already in flight rejects admission, rather than promising to undo remote effects. + +Pending closure is resumed on startup. Keep the queue and Git worktree backups +when upgrading: older plugin builds do not understand these two new statuses. +See [runtime management](runtime.md#manage-tasks-from-bot) for the UI and limits. + +While a closure is pending, the dispatcher does not start another worker pass. +An unrelated already-running task can finish; scanning continues for other tasks. +The monitor reports task maintenance until closure completes. diff --git a/docs/architecture.md b/docs/architecture.md index 4dbe9b8..e684846 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -70,3 +70,10 @@ activity snapshots. Scheduler status retains its existing RPC. The TUI polls bot independently every five seconds through the connected client; it does not infer worker activity from queue status alone. These live diagnostics do not add durable workflow phases or replace the existing ownership keepalive. + +Operator task closure is a durable dispatcher operation: `/bot` sends the owner +`automation.github.close`, which records `closing` before interruption and later +`closed`. Closed records remain as history and prevent rediscovery; scans, runtime +hooks, feedback execution and merge monitoring exclude them. Session/worktree +data is retained. See [task management](runtime.md#manage-tasks-from-bot) for +in-flight operation limits and the distinction from closing a TUI tab. diff --git a/docs/bot-workflow.md b/docs/bot-workflow.md index 2717139..3dccc85 100644 --- a/docs/bot-workflow.md +++ b/docs/bot-workflow.md @@ -9,7 +9,7 @@ published package through the release process. Read the diagrams together: sections 1–3 cover scheduling and admission, section 4 covers the saved main session, sections 5–7 cover helpers and publication, and -section 8 covers every recovery entry point. Model planning, subagents and review +section 8 covers recovery and local task closure. Model planning, subagents and review happen inside `running`; they are not additional persisted phases. ## 1. Startup, ownership, and polling @@ -32,7 +32,10 @@ flowchart TD Scan --> Save[Persist result, failures and nextAt] Save --> Clock Due -->|No| Clock - Worker --> Recover[Probe eligible stopped sessions and recover unpublished questions] + Worker --> Closing[Resume due closing requests independently of active worker] + Closing --> Clear{Any closure still pending?} + Clear -->|Yes| Worker + Clear -->|No| Recover[Probe eligible stopped sessions and recover unpublished questions] Recover --> Round[Promote one done task with pending feedback to a new round] Round --> Select[Choose ready or retry_wait task, saved running session first] Select --> Candidate{Candidate exists?} @@ -91,9 +94,9 @@ Sources: [index.ts](../src/index.ts), [easy.ts](../src/easy.ts), ```mermaid flowchart TD - Scan[Scan each configured repository] --> PRs[Refresh tracked PRs not already marked merged] - PRs --> Issues[List open issues and fetch missing tracked issues] - Issues --> Skip{PR entry or closed untracked issue?} + Scan[Scan each configured repository] --> PRs[Refresh tracked PRs excluding merged and locally closing or closed tasks] + PRs --> Issues[List open issues and fetch missing actively tracked issues] + Issues --> Skip{PR entry, locally closing or closed task, or closed untracked issue?} Skip -->|Yes| Ignore[Ignore entry] Skip -->|No| Comments[Read comments and filter authorized human comments without bot markers] Comments --> Tracked{Task already exists?} @@ -181,6 +184,10 @@ flowchart TD PB -->|Eligible retry| P Done -->|Pending authorized feedback| Round[Increment round, move feedback and reset per-round state] Round --> Q + Operator[Operator confirms Stop and close task] --> Closing[Persist closing at any saved phase] + Closing --> Drain[Interrupt known sessions and drain current operation] + Drain --> Closed[Persist closed and retain work and history] + Closing -.-> Guards[Reject new checkpoints, runtime hooks and publication] ``` Before `queued`, `analyzing`, or `commented` work advances, the dispatcher checks @@ -269,7 +276,13 @@ sequenceDiagram D->>S: Resume same session with deterministic answer message ID D->>S: Wait for completion end - alt Owner is disposed + alt Operator closes task + U->>D: Confirm Stop and close task in bot menu + D->>D: Persist closing and reject new checkpoints and prompts + D->>S: Interrupt known task sessions and wait for idleness + D->>D: Drain in-flight worker and persist closed + Note over D,G: Preserve work and history, no GitHub closure request + else Owner is disposed Note over D,S: Release local wait without interrupting healthy execution Note over D: Replacement owner loads queue and rejoins saved session else Session deadline expires @@ -363,10 +376,11 @@ flowchart TD Wait -->|Completed| Result{Succeeded outcome and non-error final assistant with finish stop?} Result -->|No| Error Result -->|Yes| Return[Return findings to main session, keep main model unchanged] + Closing[Local tracking closing or closed] --> Deny[Reject helper registration and runtime lookup] ``` Only the owning main bot session can delegate media; the helper-registration -step also requires `running` with no unresolved question. Helpers have no tools. +step also requires `running`, active tracking and no unresolved question. Helpers have no tools. URLs cannot contain credentials; local paths are resolved and must remain inside the worktree. GitHub credentials are not forwarded to media URLs. A helper uses stable session and prompt IDs for a given call. Helper failures return errors; @@ -409,6 +423,9 @@ flowchart TD Create --> Done Push --> Done Failure[Other command, model or transport error] --> Policy[Keep current phase and apply retry policy in section 8] + Close[Operator closes task before publishing starts] --> Drain[Finish in-flight local operation, reject next checkpoint] + Drain --> Preserve[Do not publish, preserve existing local changes] + InFlight[Publication already in flight] --> Refuse[Reject close request and ask operator to retry after completion] ``` Resuming `running` validates the saved session first; retrying `verifying` runs @@ -472,6 +489,9 @@ flowchart TD Manual[Manual PR close or merge] --> Refresh[Repository scan refreshes tracked PR state] Ack --> UI[Activity events and TUI polling every 10 seconds] Refresh --> UI + Local[Task closure finishes with status closed] --> UI + Menu[bot menu: select issue] --> Action[Open session, details, close tabs, restart workflow, stop and close task] + Action -->|Stop and close task| Confirm[Confirm stop and close, queue durable closing request] UI --> Busy{Associated tab busy?} Busy -->|Yes| Defer[Retry closure on a later snapshot] Busy -->|No| Tabs[Close known task and helper tabs once, preserve sessions and worktrees] @@ -520,10 +540,11 @@ Feedback after closure can still be queued, but the next round's guards block it PR-state scanning is independent of auto-merge and issue openness. The TUI subscribes to activity and polls every 10 seconds, including recovery on startup. -It opens background task tabs when enabled and exposes `/bot` for session access +It opens background task tabs when enabled and exposes `/bot` for task management and `/restartworkflow` for operator recovery in the owner project. Commands use owner-scoped RPC; they are not GitHub comment commands. Activity phases `merged` and `pr_closed` are display values, not new persisted execution phases. +Local task statuses `closing` and `closed` are durable and separate from PR state. Closure cleanup includes known earlier-round sessions and media helpers. Busy tabs wait until idle; cleanup does not delete sessions, interrupt work, or remove worktrees. A manually reopened tab is not repeatedly closed in the same TUI instance. @@ -555,6 +576,8 @@ An error normally preserves the phase so retry continues from its checkpoint. | `blocked` | Explicit `Blocked` error or GitHub HTTP 401, 404, or 422; requires inspection/retry, except a stopped session completed manually is reconciled automatically. | | `failed` | Other errors reached `maxAttempts`; operator recovery/retry required unless the checkpoint also qualifies as a stopped-session recovery candidate. | | `done` | PR publication/reconciliation completed; feedback and merge monitoring remain possible. | +| `closing` | Operator requested end of tracking; interrupt sessions and drain in-flight work, retaining errors for retry. | +| `closed` | Tracking ended locally; preserve history and work, exclude discovery, runtime hooks, execution and merge monitoring. | ```mermaid flowchart TD @@ -572,7 +595,7 @@ flowchart TD Rejoin --> Work Complete -->|No or probe fails| Retain[Retain block and feedback, probe no sooner than 30 seconds later] Retain --> Probe - Command[Operator uses restartworkflow] --> Guards{Known task, no unresolved question and no closed or merged PR?} + Command[Operator uses restartworkflow] --> Guards{Known actively tracked task, no unresolved question and no closed or merged PR?} Guards -->|No| Reject[Return actionable error, preserve checkpoint] Guards -->|Yes| Eligible{Status blocked or failed?} Eligible -->|No| Noop[accepted false, do not duplicate scheduled or completed work] @@ -589,6 +612,17 @@ flowchart TD Cancel --> Earlier[Return to commented if commentID exists, otherwise queued] Earlier --> Reset Reset --> Work + Close[bot menu: Stop and close task] --> Flight{Publication or merge already in flight?} + Flight -->|Yes| RejectClose[Reject closure, wait and try again] + Flight -->|No| SaveClose[Persist closing before interruption] + SaveClose --> Interrupt[Interrupt known sessions, missing sessions count as stopped] + Interrupt --> Drain[Wait for current worker and pending question posts] + Drain --> Again[Interrupt again to cover in-flight session creation] + Again --> Closed[Persist closed, preserve history and all local work] + Interrupt -->|Failure| CloseError[Retain closing with error, retry after 30 seconds] + Again -->|Failure| CloseError + CloseError --> Interrupt + Restart[Owner restart with saved closing request] --> Interrupt ``` - Task backoff is `min(3600, 5 * 2^attempts)` seconds, with the incremented @@ -640,6 +674,8 @@ service, resume a paused scheduler, or perform a scan itself. | Action | Saved phase and session | Effect | | --- | --- | --- | | Continue a stopped session in the TUI | Same session, `running` phase | Once successful and recognized by the probe, normal session validation, checks and publication resume automatically. | +| `/bot`, select an issue, then Stop and close task | Keep phase, sessions, worktree, branch and PR | Persist closing, interrupt saved sessions and drain work, then close local tracking. No GitHub issue/PR close or deletion. | +| `/bot`, select an issue, then Close session tabs | No checkpoint change | Close idle local tabs only, continue tracking. | | `/restartworkflow`, then select an issue | Same phase, session, worktree, branch and PR | Queue recovery for an eligible blocked/failed task. A stopped session may receive one continuation; verification/publication retries its saved stage. | | `opencode2-automation restartworkflow 'owner/repository#123'` | Same as the TUI command | Calls `automation.github.restartworkflow` with `{ key }`, returning `{ accepted }`. | | `opencode2-automation retry 'owner/repository#123'` | Same saved phase and session | Clear blocked/failed status while worker and maintenance are idle; it does not send a continuation merely because a session was stopped. | @@ -657,3 +693,29 @@ separate durable checkpoints. A failed test is never treated as session success. Regression evidence: [core.test.ts](../test/core.test.ts), [executor.test.ts](../test/executor.test.ts), [runtime.test.ts](../test/runtime.test.ts), [lifecycle.test.ts](../test/lifecycle.test.ts), [ui.test.ts](../test/ui.test.ts). + +### Local closure and missing GitHub objects + +`/bot` also exposes the saved error and task identity before any operator action. +Closing is independent of GitHub availability, issue state, PR state, route validity +and pending questions. The durable `closed` record prevents the same issue key +from being rediscovered. Scans skip closing/closed records before PR, missing-issue +and comment reads; late checkpoints and errors cannot reactivate them. There is +no automatic deletion based on an ambiguous GitHub 404 response. + +Closing preserves pending feedback and questions as history, but does not process +them. The main runtime and media helper registration reject further task activity. +The task's in-flight worker operation may finish local work before closure completes; +no subsequent verification/publication phase starts. Already-started publication +or merge refuses closure admission. In-flight comments cannot be recalled. Errors +while stopping sessions remain visible as `closing`, retried after 30 seconds or +from the menu. `accepted` acknowledges the request, not finished interruption. + +The sidebar names up to three blocked/failed/closing tasks with their saved errors, +excludes locally closed tasks from live queue counts, and shows a separate closing +count. `/bot` retains all task records and their actions, including opening the +saved conversation after closure. See [runtime management](runtime.md#manage-tasks-from-bot). + +While a closure is pending, the dispatcher does not start another worker pass. +An unrelated already-running task can finish; scanning continues for other tasks. +The monitor reports task maintenance until closure completes. diff --git a/docs/installation.md b/docs/installation.md index 3f0dd5c..4ae55c5 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -119,7 +119,7 @@ This installation procedure does not migrate sessions, queues, or worktrees. Restart the service only when work is idle. Then activate every configured owner again as shown in the README. Reopen TUI clients after UI updates to register new -commands such as `/restartworkflow`; merely reopening an old task tab does not +commands and action menus such as `/bot` task closure and `/restartworkflow`; merely reopening an old task tab does not reload its client's command registrations. A service restart preserves queue blocks and pending questions. Use [workflow recovery](runtime.md#interrupted-sessions-and-workflow-recovery) for an execution stop instead of reinstalling or deleting state. diff --git a/docs/runtime.md b/docs/runtime.md index 5dda227..738b918 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -216,6 +216,45 @@ Pausing stops scheduled scans; it does not cancel accepted tasks or active sessi Do not run independent bots on two machines against the same issues: they do not share queue ownership across machines. +## Manage tasks from /bot + +Run `/bot` in the owner project's TUI, choose an issue, then choose an action: + +- **Open session**: inspect its saved conversation, including a locally closed task. +- **Show details**: read the saved status, phase, error, branch, worktree, session, + PR link and queued feedback. These are stored checkpoints, not a fresh GitHub lookup. +- **Close session tabs**: hide that task's idle tabs in this TUI only. Busy tabs + remain open. Tracking and execution continue. +- **Restart workflow**: request the same guarded recovery as `/restartworkflow`. +- **Stop and close task**: after confirmation, persist `closing`, interrupt known + main, earlier-round and media sessions, wait for idleness and the task's in-flight + worker operation, then persist `closed`. The menu offers **Retry closing task** + while closure is pending. + +Closing tracking works for queued, waiting, failed, blocked, running and published +work, even if its GitHub issue/PR or saved OpenCode session no longer exists. +It makes no GitHub close/delete request and preserves files, branches, worktrees, +commits, session history, pending questions and feedback. It does not publish +unfinished work. `closed` here means **local tracking ended**, not PR closure. +Closed tasks remain listed as history and their conversations can be reopened. + +The closed record prevents rediscovery and later comments from restarting the +same issue. Recovery/retry cannot reopen tracking; create a new issue for new +bot work. Runtime question/helper admission is disabled once closure is requested. +Related idle tabs close once when the task becomes `closed`. + +An already-started publication or automatic merge rejects closure with an explicit +message: wait for it to finish and try again. Other in-flight operations (such as +analysis, worktree preparation or checks) may finish locally before closure +completes, but cannot advance to publication. An already-submitted GitHub comment +may complete. Closing is not a rollback of earlier Git or GitHub effects. + +Interruption errors keep the task in `closing` with a visible error, retried after +30 seconds or through **Retry closing task**. A restart resumes the saved closure +instead of restarting implementation. No success is reported while interruption +has failed or the task's worker operation is still pending. Missing sessions are +already stopped and do not block closure. + ## Runtime status sidebar The **BOT RUNTIME** section is appended to the existing right sidebar, preserving @@ -231,7 +270,9 @@ The panel shows: - Scheduler jobs: running, paused, next run time or retry delay, and errors. Pausing polling can coexist with an already-running scan or task. - Queue counts: ready/retry-wait excluding the active task, waiting for replies, blocked/failed and published - tasks, excluding closed/merged PRs. Scheduled does not mean a model is executing. + tasks, excluding closed/merged PRs and locally closed tracking. Scheduled does not mean a model is executing. + Up to three attention rows identify blocked, failed or closing issue keys and + saved errors. Pending closures have a separate count; use `/bot` for the full list. - Task details: issue, phase, round, observed main-session status, task/base branches, model, queued feedback, allocated media helper count for the current session, failed attempts, recovery request, PR state and any task/merge error. @@ -254,7 +295,7 @@ visible. Initial/unavailable data is never presented as a healthy idle service. Use `/botstatus` for a text report, including every known task and scheduler job, when the sidebar is hidden or more detail is needed. The sidebar shows up to three -scheduler jobs and truncates long labels/errors. `/bot` opens task sessions; +scheduler jobs and truncates long labels/errors. `/bot` manages task sessions; `/restartworkflow` remains the separate explicit recovery action. The TUI and owner plugin must both contain the monitor API. With an older server, @@ -311,3 +352,7 @@ identity, so use it only after inspecting the session and uncertain prompt results. `/restartworkflow` retains the session. Neither recovery command replaces service startup or scheduler `resume`. The slash command belongs in OpenCode's TUI, not in an issue comment. + +While a closure is pending, the dispatcher does not start another worker pass. +An unrelated already-running task can finish; scanning continues for other tasks. +The monitor reports task maintenance until closure completes. diff --git a/src/activity.ts b/src/activity.ts index 1299763..f777e39 100644 --- a/src/activity.ts +++ b/src/activity.ts @@ -7,6 +7,7 @@ export const Activity = z.object({ worktree: z.string().optional(), sessionReady: z.boolean(), error: z.string().optional(), prURL: z.string().optional(), prState: z.string().optional(), sessionIDs: z.array(z.string()).optional(), + closeRequestedAt: z.number().optional(), closedAt: z.number().optional(), branch: z.string().optional(), baseBranch: z.string().optional(), model: z.string().optional(), attempts: z.number().optional(), nextAt: z.number().optional(), pendingFeedback: z.number().optional(), helpers: z.number().optional(), recovery: z.boolean().optional(), @@ -21,7 +22,9 @@ export function activityOf(task: Task): Activity { ...(task.sessionID ? { sessionID: task.sessionID } : {}), ...(task.worktree ? { worktree: task.worktree } : {}), sessionReady: task.sessionReady ?? Boolean(task.promptAttempted), - ...(task.error || task.mergeError ? { error: task.error ?? task.mergeError } : {}), + ...(task.closeError || task.error || task.mergeError ? { error: task.closeError ?? task.error ?? task.mergeError } : {}), + ...(task.closeRequestedAt !== undefined ? { closeRequestedAt: task.closeRequestedAt } : {}), + ...(task.closedAt !== undefined ? { closedAt: task.closedAt } : {}), ...(task.branch ? { branch: task.branch } : {}), ...(task.baseBranch ? { baseBranch: task.baseBranch } : {}), ...(task.route ? { model: `${task.route.model.providerID}/${task.route.model.id}` } : {}), diff --git a/src/dispatcher.ts b/src/dispatcher.ts index c7fceb2..91bbbd7 100644 --- a/src/dispatcher.ts +++ b/src/dispatcher.ts @@ -14,8 +14,9 @@ export const PendingQuestion = z.object({ id: z.string(), text: z.string(), sess const Phase = z.enum(["queued", "analyzing", "commented", "running", "verifying", "publishing", "pr_opened"]); export const Task = z.object({ key: z.string(), repo: z.string(), issue: Issue, route: Route.optional(), - phase: Phase, status: z.enum(["ready", "retry_wait", "blocked", "failed", "done", "waiting"]), + phase: Phase, status: z.enum(["ready", "retry_wait", "blocked", "failed", "done", "waiting", "closing", "closed"]), attempts: z.number(), nextAt: z.number(), createdAt: z.number(), + closeRequestedAt: z.number().optional(), closedAt: z.number().optional(), closeError: z.string().optional(), analysis: z.string().optional(), commentID: z.number().optional(), analysisDecision: AnalysisDecision.optional(), analysisDialogue: z.array(z.object({ question: z.string(), answer: Comment })).optional(), @@ -43,6 +44,9 @@ export type Queue = z.infer; export class Blocked extends Error {} export class SessionStopped extends Blocked {} export class WaitingForAnswer extends Error {} +class TaskClosed extends Error {} +const closing = (task: Task) => task.status === "closing" || task.status === "closed"; +function requireTracked(task: Task) { if (closing(task)) throw new TaskClosed("Task tracking has been closed"); } function stoppedSession(task: Task) { // Recognize checkpoints from releases before sessionStopped was persisted. @@ -71,7 +75,7 @@ export interface Executor { run(task: Task, checkpoint: (patch: Partial) => Promise): Promise; verify(task: Task, repo: Repository): Promise<{ checks: string[]; commit: string }>; push(task: Task, repo: Repository): Promise; - cancel(task: Task): Promise; + cancel(task: Task, related?: boolean): Promise; completed?(task: Task): Promise; } @@ -86,6 +90,8 @@ export class Dispatcher { private lastScanStarted?: number; private lastScanFinished?: number; private scanError?: string; + private closures = new Map>(); + private publishing = new Set(); private questionPosts = new Map>(); constructor(private options: GithubOptions, private store: Store, private github: GithubPort, private executor: Executor, private signal: AbortSignal, private secrets: string[] = [], private now = Date.now, private notify: (activity: Activity) => Promise = async () => {}) {} async init() { this.queue = await this.store.load(); } @@ -93,7 +99,7 @@ export class Dispatcher { activity() { return this.queue.tasks.map(activityOf); } monitor(): DispatcherMonitor { return { ownerDirectory: this.options.ownerDirectory, - worker: this.signal.aborted ? "stopped" : this.maintenance ? "maintenance" : this.workerState, + worker: this.signal.aborted ? "stopped" : this.maintenance || this.queue.tasks.some(t => t.status === "closing") ? "maintenance" : this.workerState, scanning: Boolean(this.scanning), tasks: this.activity(), ...(this.activeTask ? { activeTask: this.activeTask } : {}), ...(this.lastScanStarted !== undefined ? { lastScanStarted: this.lastScanStarted } : {}), @@ -106,6 +112,7 @@ export class Dispatcher { let announce = false; await this.serial.run(async () => { this.signal.throwIfAborted(); + requireTracked(task); announce = Boolean(patch.sessionReady && !task.sessionReady) || Boolean(patch.status && patch.status !== task.status && ["done", "blocked", "failed", "waiting"].includes(patch.status)); announce ||= patch.pr?.state === "closed" && task.pr?.state !== "closed"; if (patch.sessionID) task.sessionIDs = [...new Set([...task.sessionIDs ?? [], ...[task.previousSessionID, task.sessionID, patch.sessionID].filter((id): id is string => Boolean(id))])]; @@ -127,13 +134,16 @@ export class Dispatcher { for (const repo of this.options.repositories) { // Watch PR state independently of automatic merging, issue state, and // worker progress so manual closure/merge also reaches attached TUIs. - for (const task of this.queue.tasks.filter(t => t.repo === repo.repo && t.pr && !t.merged)) { - const pr = await this.github.pull(repo.repo, task.pr!.number); - const merged = pr.merged === true || Boolean(pr.merged_at); - if (pr.state !== task.pr!.state || merged) await this.update(task, { pr, ...(merged ? { merged: true } : {}) }); + for (const task of this.queue.tasks.filter(t => t.repo === repo.repo && !closing(t) && t.pr && !t.merged)) { + if (closing(task)) continue; + try { + const pr = await this.github.pull(repo.repo, task.pr!.number); + const merged = pr.merged === true || Boolean(pr.merged_at); + if (pr.state !== task.pr!.state || merged) await this.update(task, { pr, ...(merged ? { merged: true } : {}) }); + } catch (error) { if (!closing(task)) throw error; } } const issues = await this.github.issues(repo.repo); - for (const tracked of this.queue.tasks.filter(t => t.repo === repo.repo)) { + for (const tracked of this.queue.tasks.filter(t => t.repo === repo.repo && !closing(t))) { if (!issues.some(i => i.number === tracked.issue.number)) issues.push(await this.github.issue(repo.repo, tracked.issue.number)); } for (const issue of issues) { @@ -141,6 +151,7 @@ export class Dispatcher { if (issue.pull_request) { ignored++; continue; } const key = `${repo.repo.toLowerCase()}#${issue.number}`; const existing = this.queue.tasks.find(t => t.key === key); + if (existing && closing(existing)) { ignored++; continue; } if (!existing && issue.state !== "open") { ignored++; continue; } const comments = await this.github.comments(repo.repo, issue.number); // A person may share the posting account with the bot. Exclude marked @@ -150,6 +161,7 @@ export class Dispatcher { if (existing) { // For queues from older versions, comments after the bot's acknowledgement are new feedback. await this.serial.run(async () => { + if (closing(existing)) return; const previousCursor = existing.commentCursor ?? existing.commentID ?? 0; const fresh = authorized.filter(c => c.id > previousCursor); let remaining = fresh; @@ -196,10 +208,12 @@ export class Dispatcher { } private authorized(login: string, authors: string[]) { return authors.some(a => a.toLowerCase() === login.toLowerCase()); } tick(): Promise { + for (const task of this.queue.tasks.filter(t => t.status === "closing" && t.nextAt <= this.now())) this.startClosing(task); if (this.maintenance) return Promise.resolve(); if (this.working) return this.working; + if (this.queue.tasks.some(t => t.status === "closing")) return Promise.resolve(); this.workerState = "reconciling"; - this.working = this.workOnce().finally(() => { this.working = undefined; this.workerState = "idle"; this.activeTask = undefined; }); + this.working = this.workOnce().catch(error => { if (!(error instanceof TaskClosed)) throw error; }).finally(() => { this.working = undefined; this.workerState = "idle"; this.activeTask = undefined; }); return this.working; } private async workOnce() { @@ -267,6 +281,7 @@ export class Dispatcher { await this.resolveAnalysis(task, repo); } if (task.phase === "commented") { + requireTracked(task); // Saved pre-upgrade analyses had no decision. Reassess them before any // implementation, preserving an already-pending base question first. if (task.question?.purpose === "base") await this.resolveBase(task, repo); @@ -275,19 +290,24 @@ export class Dispatcher { if (!task.commentID) throw new Blocked("Missing confirmed analysis comment"); if (!task.baseBranch) await this.resolveBase(task, repo); repo = { ...repo, baseBranch: task.baseBranch! }; + requireTracked(task); const workspace = await this.executor.prepare(task, repo); await this.update(task, { ...workspace, phase: "running", attempts: 0 }); } if (task.phase === "running") { + requireTracked(task); await this.executor.run(task, patch => this.update(task, patch)); if (task.question && !task.question.delivered) throw new WaitingForAnswer("Waiting for a reply in the GitHub issue"); await this.update(task, { phase: "verifying", attempts: 0, sessionStopped: undefined, recovery: undefined }); } if (task.phase === "verifying") { + requireTracked(task); const result = await this.executor.verify(task, repo); await this.update(task, { ...result, phase: "publishing", attempts: 0 }); } if (task.phase === "publishing") { + requireTracked(task); + this.publishing.add(task.key); let pr = await this.github.findPull(task.repo, task.branch); if (followup && (!pr || pr.state !== "open")) throw new Blocked("The original PR is no longer open; changes remain in the worktree"); if (followup && pr) await this.executor.push(task, repo); @@ -300,12 +320,12 @@ export class Dispatcher { await this.update(task, { pr, publishedAt: this.now(), phase: "pr_opened", status: "done", attempts: 0 }); } } catch (error) { - if (this.signal.aborted) return; + if (this.signal.aborted || closing(task) || error instanceof TaskClosed) return; if (error instanceof WaitingForAnswer) { await this.update(task, { status: task.question?.answer ? "ready" : "waiting", error: undefined }); return; } const attempts = task.attempts + 1; const blocked = error instanceof Blocked || error instanceof GithubError && [401, 404, 422].includes(error.status); await this.update(task, { attempts, sessionStopped: error instanceof SessionStopped, error: redact(error, this.secrets), status: blocked ? "blocked" : attempts >= this.options.maxAttempts ? "failed" : "retry_wait", nextAt: Math.max(this.now() + Math.min(3600, 5 * 2 ** attempts) * 1000, error instanceof GithubError ? error.retryAt ?? 0 : 0) }); - } + } finally { this.publishing.delete(task.key); } } private async resolveAnalysis(task: Task, repo: Repository) { await this.update(task, { phase: "analyzing" }); @@ -333,6 +353,7 @@ export class Dispatcher { } const lastAnswer = task.analysisDialogue?.at(-1)?.answer.id; const marker = ``; + requireTracked(task); const commentID = await this.github.ensureComment(task.repo, task.issue.number, marker, decision.comment); await this.update(task, { commentID, phase: "commented", attempts: 0, ...(task.question?.purpose === "analysis" ? { question: undefined } : {}) }); } @@ -370,7 +391,8 @@ export class Dispatcher { if (!task.publishedAt) { await this.update(task, { publishedAt: this.now() }); continue; } try { await this.scan(); // Pick up issue feedback before considering a completed task for merge. - if (task.pendingFeedback?.length || task.pr.state === "closed") continue; + if (closing(task) || task.pendingFeedback?.length || task.pr.state === "closed") continue; + this.publishing.add(task.key); const merged = await this.github.mergeApproved(task.repo, task.pr.number, task.commit, task.publishedAt, repo.allowedAuthors, this.options.autoMerge); if (merged) { await this.github.ensureComment(task.repo, task.pr.number, ``, "Pull request merged."); @@ -380,11 +402,12 @@ export class Dispatcher { if (this.signal.aborted) return; await this.update(task, { mergeError: redact(error, this.secrets), mergeNextAt: Math.max(this.now() + 60_000, error instanceof GithubError ? error.retryAt ?? 0 : 0) }); } + finally { this.publishing.delete(task.key); } } } runtime(sessionID: string) { const task = this.queue.tasks.find(t => t.sessionID === sessionID || t.helpers?.some(h => h.id === sessionID && h.parentID === t.sessionID)); - if (!task) return null; + if (!task || closing(task)) return null; const result = JSON.parse(JSON.stringify(task)) as Task; const matching = Object.values(this.options.routes).filter(r => r.agent === task.route?.agent && r.model.id === task.route?.model.id && r.model.providerID === task.route?.model.providerID); const configured = matching.length === 1 ? matching[0] : undefined; @@ -399,6 +422,7 @@ export class Dispatcher { private async askTask(task: Task, input: z.infer) { let question!: z.infer; await this.serial.run(async () => { + requireTracked(task); if (!task.question || task.question.delivered) task.question = input; question = task.question; await this.store.save(this.queue); }); @@ -406,6 +430,7 @@ export class Dispatcher { return { id: question.id }; } private async publishQuestion(task: Task, question: z.infer) { + requireTracked(task); if (!question.commentID) { const body = `Question (${question.id})\n\n${question.text}\n\n${question.permission ? `Reply with /allow ${question.id} or /deny ${question.id}.` : "Reply in this issue to continue. Only configured authors can answer."}`; const marker = ``; @@ -422,9 +447,10 @@ export class Dispatcher { } async helper(sessionID: string, callID: string, capability: "vision" | "audio") { const task = this.queue.tasks.find(t => t.sessionID === sessionID); - if (!task || task.phase !== "running" || task.question && !task.question.delivered) throw new Error("No active main bot session available for delegation"); + if (!task || closing(task) || task.phase !== "running" || task.question && !task.question.delivered) throw new Error("No active main bot session available for delegation"); const id = `ses_${createHash("sha256").update(`${sessionID}:${callID}`).digest("hex").slice(0, 32)}`; await this.serial.run(async () => { + requireTracked(task); if (!task.helpers?.some(h => h.id === id)) task.helpers = [...task.helpers ?? [], { id, parentID: sessionID, capability }]; await this.store.save(this.queue); }); @@ -441,6 +467,7 @@ export class Dispatcher { this.signal.throwIfAborted(); const task = this.queue.tasks.find(t => t.key === key); if (!task) throw new Error("Task not found in this project"); + if (closing(task)) throw new Error("Task tracking is closed; inspect its saved session or create a new issue"); if (task.question && !task.question.delivered) throw new Error("Answer the pending question or permission request in the GitHub issue first"); if (task.merged || task.pr?.state === "closed") throw new Error("The original PR is closed or merged; reopen it or create a new issue"); if (!["blocked", "failed"].includes(task.status)) return false; @@ -468,5 +495,49 @@ export class Dispatcher { await this.update(task, { status: "ready", attempts: 0, nextAt: this.now(), error: undefined }); return true; } - async settle() { await Promise.allSettled([this.scanning, this.working, this.maintenance]); } + async closeTask(key: string) { + const task = await this.serial.run(async () => { + this.signal.throwIfAborted(); + const task = this.queue.tasks.find(t => t.key === key); + if (!task) throw new Error("Task not found in this project"); + if (task.status === "closed") return undefined; + if (this.publishing.has(key)) throw new Error("Publication or merge is already in flight. Wait for it to finish, then close the task."); + Object.assign(task, { status: "closing", closeRequestedAt: task.closeRequestedAt ?? this.now(), closeError: undefined, nextAt: this.now() }); + await this.store.save(this.queue); + return task; + }); + if (!task) return false; + await this.notify(activityOf(task)).catch(() => {}); + this.startClosing(task); + return true; + } + private startClosing(task: Task) { + if (this.closures.has(task.key) || this.signal.aborted) return; + const work = this.activeTask === task.key ? this.working : undefined; + const operation = (async () => { + try { + // The durable status is saved before interruption. A replacement owner + // resumes this operation and never retries implementation/publication. + await this.executor.cancel(task, true); + await work; + await Promise.allSettled([...this.questionPosts].filter(([key]) => key.startsWith(`