From 3bee231da6a6b498f872957f56805faec8c6509d Mon Sep 17 00:00:00 2001 From: Wanderingss <1624155937@qq.com> Date: Mon, 2 Nov 2026 01:46:33 +0800 Subject: [PATCH] Execution: reclaim the directories of Ended Tickets after a retention --- docs/007-architecture.md | 5 +- src/app/http.ts | 6 +- src/app/reclaim.ts | 157 +++++++++++++++++++++++++++++++++ src/app/runtime.ts | 9 +- src/app/settings-history.ts | 4 +- src/app/views/overview.ts | 7 +- src/client/execution.tsx | 10 ++- src/client/locales.ts | 10 +++ src/client/overview.tsx | 12 ++- src/client/text.ts | 8 ++ src/limits.ts | 1 + src/policy.ts | 7 +- tests/fixtures/panel-server.ts | 6 ++ tests/panel-overview.spec.tsx | 19 +++- tests/panel-settings.spec.tsx | 4 +- tests/policy.spec.ts | 2 +- tests/reclaim.spec.ts | 69 +++++++++++++++ tests/views.spec.ts | 8 ++ 18 files changed, 326 insertions(+), 18 deletions(-) create mode 100644 src/app/reclaim.ts create mode 100644 tests/reclaim.spec.ts diff --git a/docs/007-architecture.md b/docs/007-architecture.md index 6cfc8d9..7205daa 100644 --- a/docs/007-architecture.md +++ b/docs/007-architecture.md @@ -245,6 +245,7 @@ Each model request binds the Session to the current prompt revision. The revisio | Glue | Translates | | --- | --- | | `src/app/runtime.ts` | The composition root: builds the parts with the interfaces they use, registers the routes and the Source host, and starts and stops dsh-superagent | +| `src/app/reclaim.ts` | Ended Tickets' workspace directories into what may be removed, and the disk they hold into the overview's readout | | `src/app/notices.ts` | Domain events and worker reports into notices and relays for the main Session | | `src/app/tools/main.ts`, `src/app/tools/executor.ts` | Tool arguments of the main Session and the Executor Sessions into calls of the core classes, and their results into cards | | `src/app/http.ts`, `src/app/views/`, `src/app/api.ts` | HTTP requests into calls, and Store reads into the panel's views; `api.ts` types every route for the panel | @@ -316,6 +317,7 @@ Some generic modules still refer directly to GitHub or Feishu: | `src/app/runtime.ts` | The Runtime, the composition root, with its start and stop | | `src/app/setup.ts` | Setup, and `setupChanged`, whether a save changes a field setup reads | | `src/app/passes.ts` | Passes, the pass scheduler | +| `src/app/reclaim.ts` | `Reclaimer`: hourly scans of the execution root that remove an Ended Ticket's directory once it outlives the retention, and the summary the overview reports | | `src/app/outbox.ts` | The Outbox: the Engine's `Outbox` port, the receipts and the `reply` effects that send them | | `src/app/notices.ts` | The main Session's notices and relays: `eventNotice`, `reportRelay`, `resultRelay` and `followupRelay` | | `src/app/store.ts` | The Store, and `Receipt`, a message of the outbox | @@ -436,7 +438,8 @@ Some generic modules still refer directly to GitHub or Feishu: | `tests/runtime-concurrency.spec.ts` | Work that overlaps: Triggers during a pass, Tickets that change during deliveries and launches | | `tests/http.spec.ts` | The same-origin and loopback checks, the status of each kind of failure, the body limit, path parameters and queries | | `tests/settings.spec.ts` | Settings saves that race a model or account check, a Policy sync that a save outran, and the refusals of the pause, a sync and a block | -| `tests/views.spec.ts` | What the management views derive from the Store, such as the Thread a dropped message belongs to | +| `tests/views.spec.ts` | What the management views derive from the Store, such as the Thread a dropped message belongs to, and the overview's liveness and workspace readouts | +| `tests/reclaim.spec.ts` | Which Ended Tickets' directories may be removed: the retention, the execution root, an open Ticket sharing one, and the end times that cannot be read | | `tests/timeline.spec.ts` | A Thread's pages and a Trigger's neighbours by time and then id, with every received Trigger's time in one form | | `tests/migration.spec.ts` | The schema 11 migration: every stored record in its current shape, and the refusal of older schemas | | `tests/store.spec.ts` | Repositories and their working sets, in SQLite as in memory, the kv area, with every key of `KV` spelled only in `src/kv.ts` and the migration, settings history and queries | diff --git a/src/app/http.ts b/src/app/http.ts index 9a37217..2275428 100644 --- a/src/app/http.ts +++ b/src/app/http.ts @@ -22,6 +22,7 @@ import type { Store } from './store.ts' import type { TextCache } from './cache.ts' import type { LinkTarget, ViewHost } from './views/records.ts' import type { Variable } from './execution-environment.ts' +import type { Reclaimer } from './reclaim.ts' import { policyHistory } from './settings-history.ts' import { overview } from './views/overview.ts' import { settingsView } from './views/settings.ts' @@ -46,6 +47,8 @@ export interface HttpHost { readonly masterId: string /** When this process started serving, read by the overview's liveness readout. */ readonly startedAt: string + /** Workspace reclaimer, whose summary the overview reports. */ + readonly reclaimer: Pick /** Resolves once startup finishes or setup determines what is still missing. */ readonly ready: Promise /** Resolved deployment; throws if read while waiting for setup (`!configured()`). */ @@ -162,7 +165,8 @@ function routes(runtime: HttpHost): Routes { return { 'GET state': () => stateView(runtime), 'GET overview': () => overview(views(runtime), runtime.health, policyVersion(store.settings()), - executorLoad(store, runtime.executor.activeCount()), runtime.setupProblem ?? runtime.feishuProblem, runtime.startedAt), + executorLoad(store, runtime.executor.activeCount()), runtime.setupProblem ?? runtime.feishuProblem, runtime.startedAt, + runtime.reclaimer.summary), 'GET search': ({ query }) => search(views(runtime), query.q ?? '', runtime.sourceLinks), 'GET sources': () => sourcesStatus(views(runtime), runtime.health, setupState(runtime)), 'GET settings': () => settingsView(store, runtime.sources, runtime.health, setupState(runtime), runtime.machine.stateDir), diff --git a/src/app/reclaim.ts b/src/app/reclaim.ts new file mode 100644 index 0000000..12f45e8 --- /dev/null +++ b/src/app/reclaim.ts @@ -0,0 +1,157 @@ +/** + * Reclaims the workspace directories of Tickets that ended long ago. + * A Ticket's directory holds whatever its worker cloned and wrote, so it is never deleted while the Ticket is open or can + * still be resumed soon; only directories the Executor itself named under the execution root are candidates, so an operator's + * own directory (a hand-made worktree, a shared checkout) is never touched. + */ +import { readdir, rm, stat } from 'node:fs/promises' +import { basename, join, resolve, sep } from 'node:path' +import type { Ticket } from '../core/index.ts' +import { liveTicket } from '../core/index.ts' +import type { Machine } from '../deployment.ts' +import type { Store } from './store.ts' + +/** How often the reclaimer rescans the execution root, and removes what has outlived the retention. */ +const SCAN_INTERVAL_MS = 60 * 60_000 +const DAY_MS = 86_400_000 + +export interface ReclaimHost { + readonly store: Pick + readonly machine: Pick + readonly lifetime: AbortSignal + closing(): boolean +} +/** What the panel shows about the execution root: how much of it the Ended Tickets still hold, and when it last ran. */ +export interface ReclaimSummary { + /** The execution root the Executor writes workspaces under. */ + root: string + /** Days an ended Ticket keeps its directory; `0` keeps them forever. */ + afterDays: number + /** Directories under the root that belong to a Ticket, open or ended. */ + directories: number + /** Ended Tickets whose directory has outlived the retention and goes on the next reclaim. */ + candidates: number + /** Bytes those candidates hold. */ + bytes: number + /** When the reclaimer last looked, or undefined before its first scan. */ + lastScanAt?: string + /** Disks the last scan could not read, which it left alone rather than guess at. */ + unreadable: number +} +/** A directory the reclaimer may remove, with the Ticket whose workspace it is. */ +export interface ReclaimCandidate { ticketId: string; directory: string } +export interface ReclaimResult { removed: string[]; bytes: number; failed: string[] } + +/** Whether `directory` is a path inside `root`, so removing it can never reach outside the Executor's own tree. */ +export function within(root: string, directory: string): boolean { + const inside = resolve(root), target = resolve(directory) + return target.startsWith(inside.endsWith(sep) ? inside : inside + sep) +} +/** + * The directories that may be removed now: a Ticket is ended, its directory is one the Executor named under `root`, no open + * Ticket shares it, and it has outlived `afterDays`. A retention of `0` keeps every directory. + */ +export function reclaimable(tickets: readonly Ticket[], root: string, now: number, afterDays: number): ReclaimCandidate[] { + if (afterDays <= 0) return [] + const open = new Set(tickets.filter(liveTicket).flatMap(ticket => ticket.execution ? [resolve(ticket.execution.directory)] : [])) + const seen = new Set() + return tickets.flatMap(ticket => { + const execution = ticket.execution + // A Ticket the reclaimer cannot place — never accepted, or ended with no directory — holds nothing to reclaim. + if (!execution || !ticket.ended?.at || Number.isNaN(Date.parse(ticket.ended.at))) return [] + const directory = resolve(execution.directory) + if (!within(root, directory) || basename(directory) !== execution.name) return [] + if (open.has(directory) || seen.has(directory)) return [] + const age = now - Date.parse(ticket.ended.at) + if (age < afterDays * DAY_MS) return [] + seen.add(directory) + return [{ ticketId: ticket.id, directory }] + }) +} +/** Total bytes under `directory`, or `undefined` when it cannot be read. */ +async function bytesOf(directory: string): Promise { + try { + const entries = await readdir(directory, { recursive: true, withFileTypes: true }) + let total = 0 + for (const entry of entries) { + if (!entry.isFile()) continue + const file = await stat(join(entry.parentPath, entry.name)).catch(() => undefined) + if (file) total += file.size + } + return total + } catch { return undefined } +} +/** + * Scans the execution root on a timer and removes the directories of Ended Tickets past the retention. + * The summary it exposes is what the panel reads, so the page never walks the disk itself. + */ +export class Reclaimer { + private timer?: ReturnType + private stopped = false + private running?: Promise + private current: ReclaimSummary + constructor(private readonly host: ReclaimHost) { + this.current = { root: this.root(), afterDays: this.afterDays(), directories: 0, candidates: 0, bytes: 0, unreadable: 0 } + } + private root(): string { + const settings = this.host.store.settings() + return settings.executor.directory || join(this.host.machine.stateDir, 'work') + } + private afterDays(): number { return this.host.store.settings().executor.reclaimAfterDays } + /** What the last scan found; the first scan runs at startup, so a page loaded before it reports zeroes. */ + get summary(): ReclaimSummary { return this.current } + /** Scans at startup and then hourly, removing what has outlived the retention. */ + start(): void { + if (this.stopped) return + void this.scan().then(found => found.candidates ? this.reclaim() : undefined).catch(() => { /* A scan is best effort. */ }) + this.timer = setInterval(() => { + void this.scan().then(found => found.candidates ? this.reclaim() : undefined).catch(() => { /* A scan is best effort. */ }) + }, SCAN_INTERVAL_MS) + this.timer.unref?.() + } + stop(): Promise { + this.stopped = true + clearInterval(this.timer) + return this.running?.then(() => undefined, () => undefined) ?? Promise.resolve() + } + /** Counts the directories under the root and what the candidates hold, without removing anything. */ + scan(): Promise { + return this.running ??= this.scanOnce().finally(() => { this.running = undefined }) + } + private async scanOnce(): Promise { + const root = this.root(), afterDays = this.afterDays(), now = Date.now() + const tickets = this.host.store.tickets.list() + const directories = tickets.filter(ticket => ticket.execution).flatMap(ticket => ticket.execution ? [ticket.execution] : []) + const candidates = reclaimable(tickets, root, now, afterDays) + let bytes = 0, unreadable = 0 + for (const candidate of candidates) { + const size = await bytesOf(candidate.directory) + if (size === undefined) unreadable += 1 + else bytes += size + } + this.current = { root, afterDays, directories: directories.length, candidates: candidates.length, bytes, unreadable, + lastScanAt: new Date().toISOString() } + return this.current + } + /** Removes every candidate now, whatever the timer's schedule; a directory that cannot be read is left alone and reported. */ + async reclaim(): Promise { + if (this.stopped || this.host.closing() || this.host.lifetime.aborted) return { removed: [], bytes: 0, failed: [] } + await this.scan() + const root = this.root(), afterDays = this.afterDays() + const candidates = reclaimable(this.host.store.tickets.list(), root, Date.now(), afterDays) + const removed: string[] = [], failed: string[] = [] + let bytes = 0 + for (const candidate of candidates) { + // Re-check under the current settings: an operator may have changed the root or the retention while this ran. + if (!within(this.root(), candidate.directory) || this.afterDays() <= 0) continue + const size = await bytesOf(candidate.directory) + try { + await rm(candidate.directory, { recursive: true, force: true }) + removed.push(candidate.directory) + bytes += size ?? 0 + } catch { failed.push(candidate.directory) } + } + await this.scan() + return { removed, bytes, failed } + } +} diff --git a/src/app/runtime.ts b/src/app/runtime.ts index 693eb4b..3bb853b 100644 --- a/src/app/runtime.ts +++ b/src/app/runtime.ts @@ -37,6 +37,7 @@ import { pullEndNotices } from '../sources/github-pull.ts' import { HealthBoard } from '../infra/health.ts' import { Outbox } from './outbox.ts' import { Passes } from './passes.ts' +import { Reclaimer } from './reclaim.ts' import { Setup } from './setup.ts' import { feishuSettings } from '../sources/feishu.ts' import { health, type Health, type SourceContext, type SourceHost } from '../sources/host.ts' @@ -76,6 +77,7 @@ export class Runtime extends Service implements SourceHost { readonly masterId: string /** When this process started serving, reported by the overview so a restart is visible rather than inferred. */ readonly startedAt = new Date().toISOString() + readonly reclaimer: Reclaimer private readonly sourceLinks = new TextCache(200_000) private readonly stopped = new AbortController() private readonly disposers: (() => void)[] = [] @@ -158,6 +160,7 @@ export class Runtime extends Service implements SourceHost { const evaluateDue = () => { engine.evaluateDue(); evals.set(evalsHealth(store)) } const passes = this.passes = new Passes({ engine: { receive: trigger => engine.receive(trigger), evaluateDue }, effects, health: this.runtimeHealth, lifetime: this.lifetime, configured: () => setup.configured(), closing: () => this.closing() }) + this.reclaimer = new Reclaimer({ store, machine, lifetime: this.lifetime, closing: () => this.closing() }) this.sourceDeps = { store, engine, outbox, commands, sources, get ready() { return runtime.ready }, get deployment() { return setup.deployment }, configured: () => setup.configured(), get setupProblem() { return setup.problem }, get feishuProblem() { return setup.feishuProblem }, receive: trigger => passes.receive(trigger) } @@ -216,13 +219,14 @@ export class Runtime extends Service implements SourceHost { this.executor.resume() this.passes.start() this.effects.start() + this.reclaimer.start() this.runtimeHealth.set({ state: 'ready', detail: 'Orchestrator is ready', at: now() }) } private httpHost(): HttpHost { const runtime = this const { store, sources, prompts, sessions, settings, engine, executor, health, machine, masterId, ready, sourceLinks, setup, passes } = this return { context: this.ctx, store, sources, prompts, sessions, settings, engine, executor, health, machine, masterId, ready, sourceLinks, - startedAt: this.startedAt, get deployment() { return setup.deployment }, + startedAt: this.startedAt, reclaimer: this.reclaimer, get deployment() { return setup.deployment }, get setupProblem() { return setup.problem }, get feishuProblem() { return setup.feishuProblem }, configured: () => setup.configured(), links: () => runtime.links(), ticketEnvironment: ticket => runtime.ticketEnvironment(ticket), checkSetup: () => { setup.check() }, requestPass: () => { passes.request() }, models: () => this.ctx.sessionController.modelCatalog() } @@ -231,12 +235,13 @@ export class Runtime extends Service implements SourceHost { stop(): Promise { return this.stopping ??= this.shutdown() } private async shutdown(): Promise { this.closingFlag = true; const passes = this.passes.stop(); this.stopped.abort() + const reclaiming = this.reclaimer.stop() try { this.assessor.stop() const ids = [this.masterId, ...this.store.tickets.list().map(ticket => workerSessionId(ticket.id))] const agents = ids.map(id => this.ctx.agents.get(id as SessionId)).filter((agent): agent is Agent => agent !== undefined) for (const agent of agents) agent.cancel({ kind: 'user' }, { keepInbox: true }) - await Promise.allSettled([passes, this.effects.drain(), ...agents.map(agent => agent.whenIdle())]) + await Promise.allSettled([passes, reclaiming, this.effects.drain(), ...agents.map(agent => agent.whenIdle())]) // The workers' jobs and subagents stop too: their ends would start worker turns without the tool limits released below. await this.executor.shutdown() this.executor.dispose() diff --git a/src/app/settings-history.ts b/src/app/settings-history.ts index 6a4af25..c82d9b5 100644 --- a/src/app/settings-history.ts +++ b/src/app/settings-history.ts @@ -9,7 +9,7 @@ type SourceSettings = SettingsVersion['settings']['sources'][string] export type SettingsChangeCode = 'ruleAdded' | 'ruleRenamed' | 'ruleInstruction' | 'ruleEnabled' | 'ruleDisabled' | 'ruleRemoved' | 'sourceEnabled' | 'sourceDisabled' | 'sourceField' | 'sourceRowsAdded' | 'sourceRowsRemoved' | 'sourceRowsChanged' | 'sourceAdmission' | 'mode' | 'timeout' | 'evalModel' | 'executorModel' | 'maxWorkers' | 'paused' | 'resumed' | 'executionDirectory' | 'prepareCommand' - | 'blockedCommands' + | 'blockedCommands' | 'reclaimAfterDays' /** A single settings diff entry rendered by the panel using `change.` and field labels from the Source listing. */ export interface SettingsChange { code: SettingsChangeCode; items?: string[]; from?: string | number | boolean; to?: string | number | boolean; name?: string; source?: string @@ -70,6 +70,8 @@ function settingsChanges(before: SettingsVersion['settings'], after: SettingsVer if (before.paused !== after.paused) changes.push({ code: after.paused ? 'paused' : 'resumed' }) if (before.executor.directory !== after.executor.directory) changes.push({ code: 'executionDirectory', from: before.executor.directory, to: after.executor.directory }) if (before.executor.prepare !== after.executor.prepare) changes.push({ code: 'prepareCommand' }) + if (before.executor.reclaimAfterDays !== after.executor.reclaimAfterDays) changes.push({ code: 'reclaimAfterDays', + from: before.executor.reclaimAfterDays, to: after.executor.reclaimAfterDays }) if (!same(before.executor.blockedCommands, after.executor.blockedCommands)) changes.push({ code: 'blockedCommands', from: before.executor.blockedCommands.join(' '), to: after.executor.blockedCommands.join(' ') }) for (const id of new Set([...Object.keys(before.sources), ...Object.keys(after.sources)])) changes.push(...sourceChanges(id, before.sources[id], after.sources[id], sources)) return changes diff --git a/src/app/views/overview.ts b/src/app/views/overview.ts index ac78bc3..8924352 100644 --- a/src/app/views/overview.ts +++ b/src/app/views/overview.ts @@ -4,6 +4,7 @@ import type { Health } from '../../sources/host.ts' import type { SetupProblem } from '../../setup.ts' import { anyoneAllowed } from '../../policy.ts' import type { Store } from '../store.ts' +import type { ReclaimSummary } from '../reclaim.ts' import { awaitingEval, dayAgo, newestFirst, ownersOf, snapshot, threadSummary, type ViewHost } from './records.ts' import { sourceHealth, type HealthMap } from './sources.ts' import { ticketRow, type ExecutorLoad } from './tickets.ts' @@ -52,9 +53,10 @@ export function evalsHealth(store: Store, now = Date.now()): Health { * Builds the operator overview: active system problems, tickets awaiting human action, executor load, and 24-hour trigger flow. * @param setup - Setup problem to surface first (where setup stopped, or a Source paused by setup). * @param startedAt - When this process started serving, for the liveness readout. + * @param workspace - What the Ended Tickets still hold under the execution root, and when it was last measured. */ export function overview(host: ViewHost, health: HealthMap, policyVersionValue: string, load: ExecutorLoad, setup?: SetupProblem, - startedAt?: string) { + startedAt?: string, workspace?: WorkspaceSummary) { const { store, sources } = host const view = snapshot(host) const settings = store.settings() @@ -116,6 +118,7 @@ export function overview(host: ViewHost, health: HealthMap, policyVersionValue: return { problems, liveness, + workspace: workspace ?? null, person: open.filter(ticket => ticketStage(ticket).stage === 'person').map(ticket => ticketRow(view, ticket)), executor: { ...load, maxWorkers: settings.executor.maxWorkers, idle: open.filter(ticket => ticketStage(ticket).stage === 'idle').length }, flow: { received, filtered, triggers: received - filtered, judged: judgement.judged, accepted: judgement.accepted, @@ -124,3 +127,5 @@ export function overview(host: ViewHost, health: HealthMap, policyVersionValue: } } export type OverviewView = ReturnType +/** What the execution root holds for Ended Tickets, as the overview reports it. */ +export type WorkspaceSummary = ReclaimSummary diff --git a/src/client/execution.tsx b/src/client/execution.tsx index 8c01838..d72a85f 100644 --- a/src/client/execution.tsx +++ b/src/client/execution.tsx @@ -77,9 +77,11 @@ export function ExecutionSettings() { const { blocks, environment, sources } = useSettings().settings // Keep raw user text (such as separators between blocked commands) in the draft and parse only on save. const executor = blocks.executor.value - const saved = { ...executor, maxWorkers: String(executor.maxWorkers), blockedCommands: executor.blockedCommands.join(' ') } + const saved = { ...executor, maxWorkers: String(executor.maxWorkers), reclaimAfterDays: String(executor.reclaimAfterDays), + blockedCommands: executor.blockedCommands.join(' ') } const draft = useBlockDraft('executor', saved, blocks.executor.revision, - value => ({ ...value, maxWorkers: numberOf(value.maxWorkers), blockedCommands: value.blockedCommands.split(/[\s,]+/).filter(Boolean) })) + value => ({ ...value, maxWorkers: numberOf(value.maxWorkers), reclaimAfterDays: numberOf(value.reclaimAfterDays), + blockedCommands: value.blockedCommands.split(/[\s,]+/).filter(Boolean) })) const value = draft.value const edit = (next: Partial) => draft.edit({ ...value, ...next }) const prepare = useRef(null) @@ -114,6 +116,10 @@ export function ExecutionSettings() { onChange={event => edit({ prepare: event.target.value })} /> + + edit({ reclaimAfterDays: event.target.value })} /> + edit({ blockedCommands: event.target.value })} /> diff --git a/src/client/locales.ts b/src/client/locales.ts index 8b86a85..09b2283 100644 --- a/src/client/locales.ts +++ b/src/client/locales.ts @@ -137,6 +137,12 @@ const TEXT = { livenessPolling: ['在轮询', 'Polling'], livenessNotPolling: ['不在轮询', 'Not polling'], livenessNever: ['尚无', 'Never'], + workspaceTitle: ['工单目录', 'Ticket directories'], + workspaceHeld: ['已结束工单仍占', 'Held by ended Tickets'], + workspaceCandidates: ['待删除', 'To remove'], + workspaceKeep: ['保留 {days} 天', 'Kept {days} days'], + workspaceKeepForever: ['永不删除', 'Kept forever'], + workspaceDirCount: ['目录数', 'Directories'], sourceErrorHint: ['来源会在每次轮询时自动重试。持续出现时检查网络和该来源的登录状态。', 'The Source retries on every poll. If this persists, check the network and the Source\'s sign-in.'], details: ['详情 ›', 'Details ›'], lastDay: ['过去 24 小时', 'Last 24 hours'], @@ -385,6 +391,9 @@ const TEXT = { prepareCommand: ['准备命令', 'Prepare command'], prepareCommandHint: ['新工单的目录建好前，在根目录下用 sh 运行一次，执行环境的变量是它的环境变量，例如用 git worktree 把仓库检出到 $DSH_SUPERAGENT_DIR。留空就是空目录。失败时工单显示失败，并附上命令的输出。', 'Runs once with sh in the root before a new Ticket\'s directory exists, with the execution environment\'s variables as its environment, for example to check a repository out to $DSH_SUPERAGENT_DIR with git worktree. Leave it empty for an empty directory. A failure fails the Ticket with the command\'s output.'], prepareCommandEmpty: ['留空：空目录', 'Empty: an empty directory'], + reclaimAfterDays: ['工单目录保留天数', 'Keep a Ticket directory for'], + reclaimAfterDaysHint: ['工单结束后，它的目录（检出、构建产物）保留这么久，之后由编排器删除以腾出磁盘。只删执行器自己按工单短名建的目录，根目录之外的从不删；填 0 表示永不删除。', 'After a Ticket ends, its directory (checkouts, build output) is kept this long, then the orchestrator removes it to free disk. Only directories the Executor named by Ticket short name go; nothing outside the root ever does. 0 keeps every directory.'], + reclaimAfterDaysZero: ['0：永不删除', '0: never remove'], blockedCommands: ['禁止调用的命令', 'Blocked commands'], blockedCommandsHint: ['执行器 Session 调用这些命令名时直接失败，例如用你个人账号登录的 CLI。只拦按名字的调用，用绝对路径仍能运行，所以不是安全边界。gh、git、lark-cli 不能禁止。', 'Command names that fail when an Executor Session calls them, such as a CLI signed in to your own account. Only calls by name fail; an absolute path still runs, so this is not a security boundary. gh, git and lark-cli cannot be blocked.'], blockedCommandsNone: ['以空格分隔；没有', 'Separated by spaces; none'], @@ -440,6 +449,7 @@ const TEXT = { 'change.maxWorkers': ['并行工单数：{from} → {to}', 'Concurrent Tickets: {from} → {to}'], 'change.executionDirectory': ['执行目录的根目录：{from} → {to}', 'Ticket directory root: {from} → {to}'], 'change.prepareCommand': ['准备命令', 'Prepare command'], + 'change.reclaimAfterDays': ['工单目录保留天数：{from} → {to}', 'Keep a Ticket directory for: {from} → {to}'], 'change.blockedCommands': ['禁止调用的命令：{from} → {to}', 'Blocked commands: {from} → {to}'], configHistory: ['配置历史', 'Configuration history'], 'change.evalModel': ['研判模型', 'Assessor model'], diff --git a/src/client/overview.tsx b/src/client/overview.tsx index 772b93d..0faa21b 100644 --- a/src/client/overview.tsx +++ b/src/client/overview.tsx @@ -5,7 +5,7 @@ import { Button, Icon } from './controls.tsx' import { useAction, usePanel, usePanelData } from './provider.tsx' import { setupPage, settingsRoute, type Route } from './route.ts' import { TicketLine } from './rows.tsx' -import { duration, healthText, localized, reasonLabel, valueLabel, readableMarkers } from './text.ts' +import { bytes, duration, healthText, localized, reasonLabel, valueLabel, readableMarkers } from './text.ts' import { CheckSetupButton, ConfirmButton, Hint, LoadState, Section, SetupMessage } from './widgets.tsx' function ProblemLine({ problem, onChanged }: { problem: Problem; onChanged: () => Promise }) { @@ -92,6 +92,16 @@ export function OverviewPage() { {stat(t('livenessPolling'), data.liveness.polling ? t('livenessPolling') : t('livenessNotPolling'))} + {data.workspace &&
+ {/* What ended Tickets still hold is invisible in the flow counters and grows only in the disk, so it is read out here. */} +
+ {stat(t('workspaceHeld'), bytes(data.workspace.bytes, t('lang')))} + {stat(t('workspaceCandidates'), data.workspace.candidates)} + {stat(t('workspaceDirCount'), data.workspace.directories)} + {stat(t('reclaimAfterDays'), data.workspace.afterDays === 0 ? t('workspaceKeepForever') + : t('workspaceKeep', { days: data.workspace.afterDays }))} +
+
}
{/* Received and evaluated counts operate on Triggers rather than Threads or Decisions, so they do not link to filtered views. */}
diff --git a/src/client/text.ts b/src/client/text.ts index 17812d7..cec698a 100644 --- a/src/client/text.ts +++ b/src/client/text.ts @@ -66,6 +66,14 @@ export function duration(ms: number | null | undefined, lang: string): string { if (ms < 86_400_000) return relative.format(-Math.round(ms / 3_600_000), 'hour') return relative.format(-Math.round(ms / 86_400_000), 'day') } +/** A byte count as a short localized size, for readouts such as how much a directory holds. */ +export function bytes(value: number | null | undefined, lang: string): string { + if (value === null || value === undefined || !Number.isFinite(value) || value <= 0) return '0' + const units = ['B', 'KB', 'MB', 'GB', 'TB'] + const step = Math.min(Math.floor(Math.log(value) / Math.log(1024)), units.length - 1) + const size = value / 1024 ** step + return `${step === 0 || size >= 100 ? Math.round(size) : size.toFixed(1)} ${units[step]}` +} export function fullTime(value: string | undefined, lang: string): string { const date = new Date(value ?? '') return Number.isNaN(date.getTime()) ? '' : date.toLocaleString(timeLocale(lang)) diff --git a/src/limits.ts b/src/limits.ts index 75bb2b5..7aee980 100644 --- a/src/limits.ts +++ b/src/limits.ts @@ -5,6 +5,7 @@ export const LIMITS = { evalTimeoutSeconds: { min: 30, max: 600 }, maxWorkers: { min: 1, max: 16 }, prepareCommand: 4000, + reclaimAfterDays: { min: 0, max: 365 }, promptMaxChars: { min: 1000, max: 1_000_000 }, admissionContextMaxPages: { min: 1, max: 100 }, admissionContextTimeoutMs: { min: 1, max: 90_000 }, diff --git a/src/policy.ts b/src/policy.ts index b6a66cf..3e833fe 100644 --- a/src/policy.ts +++ b/src/policy.ts @@ -3,6 +3,7 @@ * Who may originate work is configured per Source in that Source's own identities; the Policy never maps a person across Sources. */ import { LIMITS } from './limits.ts' +import { isAbsolute } from 'node:path' import { z } from 'zod' import { ADMISSION_DEFAULTS, type Admission, type AdmissionCell, type AdmissionRow } from './admission.ts' import { key, type AssessmentOutcome, type Mandate, type Policy, type PolicyEffect, type SourceScope, type Strength, type Thread, type Ticket, @@ -87,13 +88,15 @@ export const EvalConfigSchema = z.object({ }).strict() const CommandNameSchema = z.string().regex(/^[A-Za-z0-9._-]+$/).refine(name => name !== '.' && name !== '..', 'Not a command name') .refine(name => !WORKER_COMMANDS.includes(name), `${WORKER_COMMANDS.join(', ')} cannot be blocked`) -/** Executor settings: worker model, concurrency, ticket directory root (empty defaults to `/work`), `prepare` command, and blocked commands. */ +/** Executor settings: worker model, concurrency, ticket directory root (empty defaults to `/work`), `prepare` command, blocked commands, and workspace retention. */ export const ExecutorConfigSchema = z.object({ model: RoleModelSchema.nullable().default(null), maxWorkers: z.number().int().min(LIMITS.maxWorkers.min).max(LIMITS.maxWorkers.max).default(3), - directory: z.string().trim().refine(path => path === '' || path.startsWith('/'), 'An absolute path, or empty for /work').default(''), + directory: z.string().trim().refine(path => path === '' || isAbsolute(path), 'An absolute path, or empty for /work').default(''), prepare: z.string().max(LIMITS.prepareCommand).default(''), blockedCommands: z.array(CommandNameSchema).default([]), + /** Days an ended Ticket keeps its workspace directory; `0` keeps every directory, so nothing is ever removed. */ + reclaimAfterDays: z.number().int().min(LIMITS.reclaimAfterDays.min).max(LIMITS.reclaimAfterDays.max).default(14), }).strict() export const SourceInstanceSchema = z.object({ enabled: z.boolean().default(true), config: z.record(z.string(), z.unknown()).default({}) }).strict() export const SettingsSchema = z.object({ diff --git a/tests/fixtures/panel-server.ts b/tests/fixtures/panel-server.ts index f2e1136..9333ee7 100644 --- a/tests/fixtures/panel-server.ts +++ b/tests/fixtures/panel-server.ts @@ -112,6 +112,8 @@ export interface PanelServer { failing: Extract['outgoing'][] /** Sources whose poll loop the fixture reports as stopped, with how long ago their last round was. */ stalled: { id: string; title: Localized; ageMs: number; intervalMs: number }[] + /** What the execution root holds for Ended Tickets; the panel renders it as a readout. */ + workspace: NonNullable['workspace']> } export let server: PanelServer @@ -204,6 +206,7 @@ const overviewView = (): ApiResponse<'GET overview'> => { // The fixture panel renders the readout, so a poll round a few seconds old stands for a running Source. liveness: { startedAt: '2026-09-22T00:00:00Z', uptimeMs: 3_600_000, lastPollAt: new Date(Date.now() - 4_000).toISOString(), pollAgeMs: 4_000, polling: true }, + workspace: server.workspace, person: server.person, executor: { running: 2, queued: 0, maxWorkers: server.settings.executor.maxWorkers, idle: 1 }, flow: { received: 12, filtered: 2, triggers: 10, judged: 5, accepted: 1, evaluating: 0, dropped: server.dropped.filter(item => item.dropped.at >= dayAgo()).length }, @@ -257,6 +260,9 @@ function initial(): PanelServer { const defaults = { revision: 0, previousRevision: null, savedAt: '', source: 'defaults' } return { health: {}, down: false, pending: 0, opened: [], requests: [], misuse: [], queued: new Set(), dropped: [], failing: [], stalled: [], + // One ended Ticket outlived the retention, so the readout has something to report. + workspace: { root: '/srv/dsh-superagent/work', afterDays: 14, directories: 5, candidates: 1, bytes: 58 * 1024 * 1024, + lastScanAt: new Date(Date.now() - 600_000).toISOString(), unreadable: 0 }, setup: { machine: MACHINE, problem: null, feishuProblem: null, accounts: { repository: 'acme/widgets', agent: 'agent-bot', feishu: { appId: 'cli_bot', brand: 'feishu', botId: 'ou_bot', name: 'Widgets Bot' } } }, diff --git a/tests/panel-overview.spec.tsx b/tests/panel-overview.spec.tsx index ac5475f..2b602db 100644 --- a/tests/panel-overview.spec.tsx +++ b/tests/panel-overview.spec.tsx @@ -11,8 +11,8 @@ it('leads the overview with the pending Tickets, then problems, the last day and await mount() expect(tab(zh['tab.overview']).getAttribute('aria-selected')).toBe('true') expect(all('[role="tab"]').map(tab => tab.textContent)).toEqual(['概览', '来源', '触发', '评估', '执行', '设置']) - expect(all('.dt-section-title').map(title => title.textContent)).toEqual(['待处理工单2', '系统问题2', '存活', '过去 24 小时详情 ›', - '执行器详情 ›']) + expect(all('.dt-section-title').map(title => title.textContent)).toEqual(['待处理工单2', '系统问题2', '存活', '工单目录', + '过去 24 小时详情 ›', '执行器详情 ›']) expect(all('.dt-cards .dt-ticket-stage').map(item => item.textContent)).toEqual(['待回答', '待处理失败']) expect(all('.dt-cards .dt-att-what').map(item => item.textContent)).toEqual(['飞书 来源异常', '评估未完成：这个文案好奇怪']) // Actions sit below the title and Threads, apart from the stage, each with its icon. @@ -22,9 +22,11 @@ it('leads the overview with the pending Tickets, then problems, the last day and expect(text()).toContain('需要确认按钮颜色') // Durations are formatted by the interface language, so the readout is asserted by label, and only the polling word verbatim. expect(all('.dt-stat-label').map(label => label.textContent)).toEqual([zh.livenessUptime, zh.livenessLastPoll, zh.livenessPolling, + zh.workspaceHeld, zh.workspaceCandidates, zh.workspaceDirCount, zh.reclaimAfterDays, zh.flowReceivedLabel, zh.flowFilteredLabel, zh.flowJudgedLabel, zh.flowAcceptedLabel, zh.executingLabel, zh.queuedLabel, zh.idleLabel]) - expect(all('.dt-stat-value').map(value => value.textContent).slice(3)).toEqual(['12', '2', '5', '1', '2 / 3', '0', '1']) + expect(all('.dt-stat-value').map(value => value.textContent).slice(7)).toEqual(['12', '2', '5', '1', '2 / 3', '0', '1']) + expect(all('.dt-stat-value').map(value => value.textContent)[3]).toBe('58.0 MB') expect(all('.dt-stat-value').map(value => value.textContent).slice(0, 3)).toEqual([expect.stringContaining('小时'), expect .stringContaining('秒'), zh.livenessPolling]) expect(all('.dt-chip').map(chip => chip.textContent)[0]).toBe('GitHub#42CI 失败1') @@ -134,9 +136,10 @@ it('words a refused retry in the interface language by its code', async () => { it('shows the overview in English', async () => { await mount(en) expect(all('.dt-section-title').map(title => title.textContent)).toEqual([`${en.pendingTickets}2`, `${en.problems}2`, - en.livenessTitle, `${en.lastDay}${en.details}`, `${en['term.executor']}${en.details}`]) + en.livenessTitle, en.workspaceTitle, `${en.lastDay}${en.details}`, `${en['term.executor']}${en.details}`]) expect(all('.dt-cards .dt-ticket-stage').map(item => item.textContent)).toEqual([en['stage.answer'], en['stage.failed']]) expect(all('.dt-stat-label').map(label => label.textContent)).toEqual([en.livenessUptime, en.livenessLastPoll, en.livenessPolling, + en.workspaceHeld, en.workspaceCandidates, en.workspaceDirCount, en.reclaimAfterDays, en.flowReceivedLabel, en.flowFilteredLabel, en.flowJudgedLabel, en.flowAcceptedLabel, en.executingLabel, en.queuedLabel, en.idleLabel]) }) @@ -174,3 +177,11 @@ it('says when a Source stopped polling, with how long its last round was and the expect(line).toBe(zh.problemStalled.replace('{source}', 'GitHub').replace('{age}', duration(26 * 60_000, 'zh'))) expect(text()).toContain(zh.problemStalledHint) }) + +it('reads out what the ended Tickets still hold, and that nothing is ever removed when the retention is zero', async () => { + server.workspace = { root: '/srv/work', afterDays: 0, directories: 3, candidates: 0, bytes: 0, lastScanAt: new Date().toISOString(), + unreadable: 0 } + await mount() + expect(text()).toContain(zh.workspaceKeepForever) + expect(all('.dt-stat-value').map(value => value.textContent)).toContain('0') +}) diff --git a/tests/panel-settings.spec.tsx b/tests/panel-settings.spec.tsx index 0208e06..ed953c4 100644 --- a/tests/panel-settings.spec.tsx +++ b/tests/panel-settings.spec.tsx @@ -54,8 +54,8 @@ it('saves the Eval block on the Policy settings and the Executor with its execut expect([...document.querySelectorAll('#dt-worker-model-efforts option')].map(option => option.getAttribute('value'))).toEqual(['high']) await click(zh.save) expect(server.settings.executor).toEqual({ maxWorkers: 5, directory: '/srv/work', - prepare: 'git worktree add "$DSH_SUPERAGENT_DIR"', blockedCommands: ['gh-personal', 'gh.other'], model: { provider: 'worker-provider', - model: 'worker-model', reasoningEffort: 'high' } }) + prepare: 'git worktree add "$DSH_SUPERAGENT_DIR"', blockedCommands: ['gh-personal', 'gh.other'], reclaimAfterDays: 14, + model: { provider: 'worker-provider', model: 'worker-model', reasoningEffort: 'high' } }) expect(server.requests.filter(request => request.method === 'PUT').map(request => request.path)).toEqual(['settings/eval', 'settings/executor']) }) diff --git a/tests/policy.spec.ts b/tests/policy.spec.ts index 4a25b14..acc26d3 100644 --- a/tests/policy.spec.ts +++ b/tests/policy.spec.ts @@ -254,7 +254,7 @@ describe('settings migration', () => { sources: { feishu: feishuSource({ chats: [{ chatId: CHAT, listen: 'mentions' }, { chatId: OBSERVED, listen: 'all' }], directMessages: false }), github: { enabled: true, config: { pollSeconds: 120 } } }, policy: { rules: [], sources: { feishu: { allowed: [{ id: MEMBER, name: 'A' }], admission: {} }, github: { allowed: [{ id: 'a' }], admission: {} } } }, eval: { mode: 'auto', timeoutSeconds: 90, model: { provider: 'p', model: 'm' } }, - executor: { model: null, maxWorkers: 5, directory: '', prepare: '', blockedCommands: [] }, + executor: { model: null, maxWorkers: 5, directory: '', prepare: '', blockedCommands: [], reclaimAfterDays: 14 }, }) expect(migrateSettings(migrated)).toEqual(migrated) }) diff --git a/tests/reclaim.spec.ts b/tests/reclaim.spec.ts new file mode 100644 index 0000000..3173756 --- /dev/null +++ b/tests/reclaim.spec.ts @@ -0,0 +1,69 @@ +// Reclaiming the directories of Ended Tickets: what may go, and what the readout counts. +import { resolve } from 'node:path' +import { expect, it } from 'vitest' +import { reclaimable, within } from '../src/app/reclaim.ts' +import type { Ticket } from '../src/core/index.ts' + +const DAY = 86_400_000 +const root = '/srv/dsh-superagent/work' +const now = Date.parse('2026-10-31T00:00:00.000Z') +const at = (daysAgo: number) => new Date(now - daysAgo * DAY).toISOString() + +/** A Ticket as the reclaimer reads it: an id, its workspace, and when it ended. */ +function ticket(id: string, name: string, ended: string | undefined, directory = `${root}/${name}`): Ticket { + return { id, title: id, decisionId: 'd', channels: [], undelivered: [], createdAt: at(30), updatedAt: at(30), + mandate: { strength: 'explicit', policyVersion: 'v', basis: { via: 'listed' }, actor: { source: 'feishu', id: 'ou_alice' }, scope: '' }, + origin: { triggerId: 'e1', triggerIds: ['e1'] }, assignments: [], inputs: [], + ...ended ? { ended: { at: ended, by: 'executor' as const, outcome: 'completed' as const } } : {}, + execution: { name, directory, assignment: '' } } +} + +it('tells a path inside the execution root from one beside or above it', () => { + expect(within(root, `${root}/create-wiki-06e6e4ae`)).toBe(true) + expect(within(root, `${root}/a/b/c`)).toBe(true) + // A sibling whose name starts with the root's, a parent, and the root itself are not inside it. + expect(within(root, `${root}-other/ticket`)).toBe(false) + expect(within(root, '/srv/dsh-superagent')).toBe(false) + expect(within(root, root)).toBe(false) + expect(within(root, '/elsewhere/ticket')).toBe(false) +}) + +it('reclaims an Ended Ticket\'s own directory once it outlives the retention, and no other', () => { + const tickets = [ticket('t1', 'create-wiki-06e6e4ae', at(20)), ticket('t2', 'create-hi-7aa4cedc44b4', at(2))] + expect(reclaimable(tickets, root, now, 14).map(item => item.ticketId)).toEqual(['t1']) + // A Ticket still open holds its directory however old it is. + expect(reclaimable([ticket('t1', 'create-wiki-06e6e4ae', undefined)], root, now, 14)).toEqual([]) + // A retention of zero keeps everything, which is how an operator opts out entirely. + expect(reclaimable(tickets, root, now, 0)).toEqual([]) + // The retention is a whole number of days, so a Ticket ended a moment ago waits. + expect(reclaimable([ticket('t1', 'create-wiki-06e6e4ae', at(13))], root, now, 14)).toEqual([]) +}) + +it('never reclaims a directory the Executor did not name under its root', () => { + // An operator's own directory outside the root: the Executor recorded it, but removing it would reach past the tree. + expect(reclaimable([ticket('t1', 'my-checkout', at(20), '/home/me/my-checkout')], root, now, 14)).toEqual([]) + // A root that merely shares a prefix with the directory is not the parent of it. + expect(reclaimable([ticket('t1', 'ticket', at(20), `${root}-other/ticket`)], root, now, 14)).toEqual([]) + // A directory the Executor recorded under a different name than the Ticket's own is not one it made. + expect(reclaimable([ticket('t1', 'create-wiki-06e6e4ae', at(20), `${root}/something-else`)], root, now, 14)).toEqual([]) + // A Ticket that never got a directory, such as one never accepted, holds nothing. + const bare: Ticket = { ...ticket('t1', 'x', at(20)), execution: undefined } + expect(reclaimable([bare], root, now, 14)).toEqual([]) +}) + +it('leaves a directory that an open Ticket uses, and lists each directory once', () => { + // A Ticket that ended is not removed while a newer open Ticket works in the same directory. + const shared = `${root}/create-wiki-06e6e4ae` + const tickets = [ticket('t1', 'create-wiki-06e6e4ae', at(20), shared), ticket('t2', 'create-wiki-06e6e4ae', undefined, shared)] + expect(reclaimable(tickets, root, now, 14)).toEqual([]) + // Two ended Tickets naming the same directory reclaim it once, not twice. + const twice = [ticket('t1', 'create-wiki-06e6e4ae', at(20), shared), ticket('t2', 'create-wiki-06e6e4ae', at(25), shared)] + // Paths come back resolved, so the expectation is resolved too rather than spelled with the platform's separator. + expect(reclaimable(twice, root, now, 14)).toEqual([{ ticketId: 't1', directory: resolve(shared) }]) + // Either order reclaims it once, named after whichever Ticket the list reaches first. + expect(reclaimable([...twice].reverse(), root, now, 14)).toEqual([{ ticketId: 't2', directory: resolve(shared) }]) +}) + +it('ignores an Ended Ticket whose end time cannot be read', () => { + expect(reclaimable([ticket('t1', 'create-wiki-06e6e4ae', 'not a time')], root, now, 14)).toEqual([]) +}) diff --git a/tests/views.spec.ts b/tests/views.spec.ts index 5665e89..bbb6ae4 100644 --- a/tests/views.spec.ts +++ b/tests/views.spec.ts @@ -355,3 +355,11 @@ it('reports whether the service is alive and polling apart from whether any Trig expect(overview(host, {}, 'v1', load, undefined, startedAt).liveness).toMatchObject({ polling: false, lastPollAt: null, pollAgeMs: null }) expect(overview(host, {}, 'v1', load).liveness).toMatchObject({ startedAt: null, uptimeMs: null, polling: false }) }) + +it('reports what the execution root holds for Ended Tickets apart from what any Trigger did', () => { + const host = views() + const workspace = { root: '/srv/work', afterDays: 14, directories: 5, candidates: 1, bytes: 1024, lastScanAt: at, unreadable: 0 } + expect(overview(host, {}, 'v1', load, undefined, undefined, workspace).workspace).toEqual(workspace) + // Before the first scan there is nothing to say, which the page renders as no readout rather than as zeroes. + expect(overview(host, {}, 'v1', load).workspace).toBeNull() +})