Skip to content

Commit c965f99

Browse files
cevrcevr
committed
feat(core): shells can be killed, backgrounded and listed with exited ones
Shell.kill sends SIGTERM to the process group and SIGKILL to what is left after 3 s, keeping the record and output. Output rotates at 8 MiB into <id>.out.1 (about 16 MiB per shell, oldest bytes dropped). Exited background shells are retained apart from foreground ones (100 vs 25). Job.background takes a running option, and experimental.shell_background_after sets when a foreground shell call moves to the background.
1 parent b7d6c4f commit c965f99

7 files changed

Lines changed: 339 additions & 85 deletions

File tree

‎packages/core/src/config/normalize.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -531,6 +531,15 @@ function normalizeExperimental(
531531
)
532532
if (value !== undefined) result.subagent_depth = value
533533
}
534+
if (own(experimental, "shell_background_after")) {
535+
const value = decodeEncoded(
536+
ConfigExperimental.Info.fields.shell_background_after,
537+
experimental.shell_background_after,
538+
["experimental", "shell_background_after"],
539+
diagnostics,
540+
)
541+
if (value !== undefined) result.shell_background_after = value
542+
}
534543
if (own(experimental, "location_idle_timeout")) {
535544
const value = decodeEncoded(
536545
ConfigExperimental.Info.fields.location_idle_timeout,

‎packages/core/src/job.ts‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -148,7 +148,9 @@ export interface Interface {
148148
readonly start: (input: StartInput) => Effect.Effect<Info>
149149
readonly wait: (input: WaitInput) => Effect.Effect<WaitResult>
150150
readonly block: (input: BlockInput) => Effect.Effect<BlockResult | undefined>
151-
readonly background: (id: string) => Effect.Effect<Info | undefined>
151+
// With `running`, a job that already settled stays foreground even when recoverable, so a timer
152+
// racing the job's own completion never turns a finished result into a background notice.
153+
readonly background: (id: string, options?: { readonly running?: boolean }) => Effect.Effect<Info | undefined>
152154
readonly backgroundAll: (input: BackgroundAllInput) => Effect.Effect<Info[]>
153155
readonly cancel: (id: string) => Effect.Effect<Info | undefined>
154156
readonly cancelSession: (sessionID: SessionSchema.ID) => Effect.Effect<Info[]>
@@ -439,13 +441,13 @@ export const make = Effect.gen(function* () {
439441
return next
440442
})
441443

442-
const background: Interface["background"] = Effect.fn("Job.background")(function* (id) {
444+
const background: Interface["background"] = Effect.fn("Job.background")(function* (id, options) {
443445
const result = yield* SynchronizedRef.modifyEffect(
444446
state.jobs,
445447
Effect.fnUntraced(function* (jobs): Effect.fn.Return<readonly [BackgroundResult, Map<string, Active>]> {
446448
const job = jobs.get(id)
447449
// Recoverable work may finish before the caller backgrounds it.
448-
if (!job || (job.info.status !== "running" && !job.recovery)) return [{}, jobs]
450+
if (!job || (job.info.status !== "running" && (!job.recovery || options?.running))) return [{}, jobs]
449451
if (job.isBackgrounded) return [{ info: snapshot(job) }, jobs]
450452
const next = yield* markBackground(job)
451453
return [{ info: snapshot(next), backgrounded: job.backgrounded }, new Map(jobs).set(id, next)]

‎packages/core/src/shell.ts‎

Lines changed: 150 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,18 @@ export class NotFoundError extends Schema.TaggedError<NotFoundError>()("Shell.No
2929

3030
// Keep recent exited processes observable in memory, including their file-backed output.
3131
// The process-local cap complements the time-based sweep, which also cleans files left by restarts.
32+
// Background shells (`metadata.background`) keep their own, larger allowance, so a run of quick
33+
// foreground commands never evicts the background work an agent may still read.
3234
const EXITED_LIMIT = 25
35+
export const BACKGROUND_EXITED_LIMIT = 100
3336
export const RETENTION = Duration.days(7)
3437
export const DIRECTORY = "shell"
38+
// Output is kept in at most two files of about this size each: `<id>.out` holds the newest bytes,
39+
// `<id>.out.1` the segment before it. A command that writes more loses its oldest output, so one
40+
// shell never holds more than about twice this on disk.
41+
export const OUTPUT_SEGMENT_BYTES = 8 * 1024 * 1024
42+
// How long a kill waits after SIGTERM before it sends SIGKILL to the process group.
43+
export const KILL_GRACE = Duration.seconds(3)
3544

3645
type Info = Shell.Info
3746
type CreateInput = Shell.CreateInput & {
@@ -44,15 +53,20 @@ type Active = {
4453
file: string
4554
size: number
4655
newlines: number
56+
// The absolute byte offset `file` starts at, and, once output rotated, where `file.1` starts.
57+
base: number
58+
previous?: number
4759
// Resolves with the terminal Info once the command exits, times out, or is killed. A wait
4860
// started after termination resolves immediately from the already-completed deferred.
4961
done: Deferred.Deferred<Info, NotFoundError>
5062
timeoutFiber?: Fiber.Fiber<void>
5163
timeout?: (duration: number) => Effect.Effect<void>
64+
// Ends the command; the fiber it returns settles once the whole process group is gone.
65+
kill?: Effect.Effect<Fiber.Fiber<void> | undefined>
5266
}
5367

5468
type Ended = {
55-
status: "exited" | "timeout"
69+
status: "exited" | "timeout" | "killed"
5670
exit?: number
5771
signal?: string
5872
completed: number
@@ -71,8 +85,8 @@ export interface Interface {
7185
input: CreateInput,
7286
before?: (input: ShellCreateBefore) => Effect.Effect<void, E, R>,
7387
) => Effect.Effect<Shell.Info, E | AppProcess.AppProcessError, R>
74-
// Currently running commands only; exited shells are retained for get/output but excluded here.
75-
readonly list: () => Effect.Effect<Shell.Info[]>
88+
// Currently running commands; with `exited`, also the retained exited ones, oldest first.
89+
readonly list: (options?: { readonly exited?: boolean }) => Effect.Effect<Shell.Info[]>
7690
readonly get: (id: Shell.ID) => Effect.Effect<Shell.Info, NotFoundError>
7791
// Resolves once the command reaches a terminal status, returning its final Info. Fails with
7892
// NotFoundError if the command is unknown or is removed before it terminates.
@@ -82,6 +96,12 @@ export interface Interface {
8296
// Replaces the running command's timeout from now; zero clears it. Ignored once the command has
8397
// ended or timed out, even while its process is still being killed or its output captured.
8498
readonly timeout: (id: Shell.ID, duration: number) => Effect.Effect<Shell.Info, NotFoundError>
99+
// Moves a running command to the background: clears its timeout and tags it `metadata.background`,
100+
// which keeps it under the background retention allowance once it exits.
101+
readonly background: (id: Shell.ID) => Effect.Effect<Shell.Info, NotFoundError>
102+
// Ends a running command: SIGTERM to its process group, SIGKILL after KILL_GRACE to whatever of the
103+
// group is left. Its record and output stay; resolves with its terminal Info (status "killed").
104+
readonly kill: (id: Shell.ID) => Effect.Effect<Shell.Info, NotFoundError>
85105
readonly output: (id: Shell.ID, input?: Shell.OutputInput) => Effect.Effect<Shell.Output, NotFoundError>
86106
readonly remove: (id: Shell.ID) => Effect.Effect<void, NotFoundError>
87107
}
@@ -102,7 +122,7 @@ export const cleanup = Effect.fn("Shell.cleanup")(function* () {
102122
fs.readDirectoryEntries(path.join(directory, project.name)).pipe(
103123
Effect.map((entries) =>
104124
entries.flatMap((entry) =>
105-
entry.type === "file" && /^sh_[0-9a-f]{12}.*\.out$/.test(entry.name)
125+
entry.type === "file" && /^sh_[0-9a-f]{12}.*\.out(\.1)?$/.test(entry.name)
106126
? [path.join(directory, project.name, entry.name)]
107127
: [],
108128
),
@@ -114,6 +134,21 @@ export const cleanup = Effect.fn("Shell.cleanup")(function* () {
114134
yield* FileRetention.cleanup(fs, files.flat(), RETENTION)
115135
})
116136

137+
const groupAlive = (pid: number) => {
138+
try {
139+
process.kill(-pid, 0)
140+
return true
141+
} catch {
142+
return false
143+
}
144+
}
145+
146+
const signalGroup = (pid: number, signal: NodeJS.Signals) => {
147+
try {
148+
process.kill(-pid, signal)
149+
} catch {}
150+
}
151+
117152
const cleanupLayer = Layer.effectDiscard(
118153
cleanup().pipe(Effect.repeat(Schedule.spaced(Duration.hours(1))), Effect.forkScoped),
119154
)
@@ -142,7 +177,7 @@ const layer = () =>
142177
const exitOrder: Shell.ID[] = []
143178

144179
const outputDir = path.join(global.data, DIRECTORY, location.project.id)
145-
const { mkdir, unlink } = yield* Effect.promise(() => import("fs/promises"))
180+
const { mkdir, unlink, rename } = yield* Effect.promise(() => import("fs/promises"))
146181
const { createWriteStream, createReadStream } = yield* Effect.promise(() => import("fs"))
147182
yield* Effect.promise(() => mkdir(outputDir, { recursive: true }))
148183

@@ -173,7 +208,9 @@ const layer = () =>
173208
if (command.timeoutFiber) yield* Fiber.interrupt(command.timeoutFiber)
174209
// Unblock any wait still pending when the command is removed before it terminated.
175210
yield* Deferred.fail(command.done, new NotFoundError({ id }))
176-
yield* Effect.promise(() => unlink(command.file).catch(() => {}))
211+
yield* Effect.promise(() =>
212+
Promise.all([unlink(command.file).catch(() => {}), unlink(`${command.file}.1`).catch(() => {})]),
213+
)
177214
// Every shell reports exactly one exit; removing a running one ends it as killed.
178215
if (command.info.status === "running") yield* bus.publish(Shell.Event.Exited, { id, status: "killed" })
179216
yield* bus.publish(Shell.Event.Deleted, { id })
@@ -184,9 +221,9 @@ const layer = () =>
184221
yield* removeCommand(id)
185222
})
186223

187-
const list = Effect.fn("Shell.list")(function* () {
224+
const list = Effect.fn("Shell.list")(function* (options?: { readonly exited?: boolean }) {
188225
return Array.from(commands.values())
189-
.filter((command) => command.info.status === "running")
226+
.filter((command) => options?.exited === true || command.info.status === "running")
190227
.map((command) => command.info)
191228
})
192229

@@ -205,33 +242,62 @@ const layer = () =>
205242
return command.info
206243
})
207244

245+
const backgroundCommand = Effect.fn("Shell.background")(function* (id: Shell.ID) {
246+
const command = yield* require(id)
247+
if (command.info.status !== "running") return command.info
248+
if (command.timeout) yield* command.timeout(0)
249+
command.info = produce(command.info, (draft) => {
250+
draft.metadata = { ...draft.metadata, background: true }
251+
})
252+
return command.info
253+
})
254+
255+
const kill = Effect.fn("Shell.kill")(function* (id: Shell.ID) {
256+
const command = yield* require(id)
257+
const sweep = command.info.status === "running" && command.kill ? yield* command.kill : undefined
258+
const info = yield* Deferred.await(command.done)
259+
if (sweep) yield* Fiber.join(sweep)
260+
return info
261+
})
262+
263+
// Reads `length` bytes of one segment file from its own offset `start`.
264+
const readFile = (file: string, start: number, length: number) =>
265+
Effect.promise(
266+
() =>
267+
new Promise<Buffer>((resolve) => {
268+
if (length <= 0) return resolve(Buffer.alloc(0))
269+
const chunks: Buffer[] = []
270+
const stream = createReadStream(file, { start, end: start + length - 1 })
271+
stream.on("data", (chunk: string | Buffer) => chunks.push(Buffer.from(chunk)))
272+
stream.on("end", () => resolve(Buffer.concat(chunks)))
273+
stream.on("error", () => resolve(Buffer.concat(chunks)))
274+
}),
275+
)
276+
208277
const output = Effect.fnUntraced(function* (id: Shell.ID, input?: Shell.OutputInput) {
209278
const command = yield* require(id)
210279
const cursor = input?.cursor ?? 0
211280
const limit = input?.limit ?? 65536
212281
if (cursor >= command.size) return { output: "", cursor: command.size, size: command.size, truncated: false }
213-
const start = Math.max(0, cursor)
282+
// Bytes before the oldest kept segment were dropped by rotation; a read from there starts at the oldest kept.
283+
const earliest = command.previous ?? command.base
284+
const start = Math.max(earliest, cursor)
214285
const length = Math.min(limit, command.size - start)
215-
const buffer = Buffer.alloc(length)
216-
const bytesRead = yield* Effect.promise(
217-
() =>
218-
new Promise<number>((resolve) => {
219-
const stream = createReadStream(command.file, { start, end: start + length - 1 })
220-
let offset = 0
221-
stream.on("data", (chunk: string | Buffer) => {
222-
const bytes = Buffer.from(chunk)
223-
bytes.copy(buffer, offset)
224-
offset += bytes.length
225-
})
226-
stream.on("end", () => resolve(offset))
227-
stream.on("error", () => resolve(0))
228-
}),
229-
)
286+
const base = command.base
287+
const older =
288+
command.previous !== undefined && start < base
289+
? yield* readFile(`${command.file}.1`, start - command.previous, Math.min(length, base - start))
290+
: Buffer.alloc(0)
291+
const newer =
292+
start + length > base
293+
? yield* readFile(command.file, Math.max(0, start - base), start + length - Math.max(start, base))
294+
: Buffer.alloc(0)
295+
const bytes = Buffer.concat([older, newer])
230296
return {
231-
output: buffer.subarray(0, bytesRead).toString("utf8"),
232-
cursor: start + bytesRead,
297+
output: bytes.toString("utf8"),
298+
cursor: start + bytes.length,
233299
size: command.size,
234-
truncated: false,
300+
truncated: cursor < earliest,
235301
}
236302
})
237303

@@ -251,11 +317,15 @@ const layer = () =>
251317
if (page.output.endsWith("\n")) lines.pop()
252318
const truncated = latest.size > maxBytes || lines.length > maxLines
253319
const text = lines.length > maxLines ? lines.slice(-maxLines).join("\n") : page.output
254-
const total = (yield* require(info.id)).newlines + (page.output.endsWith("\n") ? 0 : 1)
320+
const command = yield* require(info.id)
321+
const total = command.newlines + (page.output.endsWith("\n") ? 0 : 1)
255322
const shown = Math.min(lines.length, maxLines)
256-
const notice = truncated
257-
? `\n\n[showing lines ${total - shown + 1}-${total} of ${total}; full output saved to ${info.file}]`
258-
: ""
323+
const kept = command.previous ?? command.base
324+
const notice = !truncated
325+
? ""
326+
: kept > 0
327+
? `\n\n[showing lines ${total - shown + 1}-${total} of ${total}; the newest output is saved to ${info.file}, the first ${kept} bytes were dropped]`
328+
: `\n\n[showing lines ${total - shown + 1}-${total} of ${total}; full output saved to ${info.file}]`
259329
return { output: `${text || "(no output)"}${notice}`, truncated }
260330
}).pipe(Effect.catchTag("Shell.NotFoundError", () => Effect.succeed(undefined)))
261331
return { info, capture }
@@ -326,15 +396,28 @@ const layer = () =>
326396
file,
327397
size: 0,
328398
newlines: 0,
399+
base: 0,
329400
done: Deferred.makeUnsafe<Info, NotFoundError>(),
330401
}
331402
commands.set(id, command)
332403

333-
const stream = createWriteStream(file)
404+
let stream = createWriteStream(file)
334405
const outputDone = Latch.makeUnsafe()
335406
const pump = handle.all.pipe(
336407
Stream.runForEach((chunk: Uint8Array) =>
337-
Effect.sync(() => {
408+
Effect.gen(function* () {
409+
// Rotate: the full segment becomes `<id>.out.1` (replacing the one before it) and a fresh
410+
// `<id>.out` starts at the current absolute offset.
411+
if (command.size > command.base && command.size - command.base + chunk.length > OUTPUT_SEGMENT_BYTES) {
412+
const old = stream
413+
yield* Effect.promise(async () => {
414+
await new Promise<void>((resolve) => old.end(() => resolve()))
415+
await rename(file, `${file}.1`).catch(() => {})
416+
})
417+
command.previous = command.base
418+
command.base = command.size
419+
stream = createWriteStream(file)
420+
}
338421
stream.write(chunk)
339422
command.size += chunk.length
340423
// Count while streaming so truncation notices never rescan the output file.
@@ -392,6 +475,22 @@ const layer = () =>
392475

393476
yield* command.timeout(invocation.timeout)
394477

478+
command.kill = Effect.gen(function* () {
479+
if (!(yield* end({ status: "killed", signal: "SIGTERM" }))) return undefined
480+
// Detached like the timeout kill, so an interrupted caller never cancels it halfway.
481+
return runFork(
482+
Effect.gen(function* () {
483+
yield* handle.kill({ forceKillAfter: KILL_GRACE }).pipe(Effect.catch(() => Effect.void))
484+
// The spawner waits on the group leader only; a child that ignores SIGTERM outlives it
485+
// and keeps the output pipe open. Sweep the group once more after the grace.
486+
const pid = handle.pid
487+
if (process.platform === "win32" || !pid || !groupAlive(pid)) return
488+
yield* Effect.sleep(KILL_GRACE)
489+
if (groupAlive(pid)) signalGroup(pid, "SIGKILL")
490+
}),
491+
)
492+
}).pipe(Effect.uninterruptible)
493+
395494
// The process exits even after a timeout kill, so this fiber always publishes the outcome.
396495
runFork(
397496
Effect.gen(function* () {
@@ -426,10 +525,12 @@ const layer = () =>
426525
status: outcome.status,
427526
})
428527
exitOrder.push(id)
429-
while (exitOrder.length > EXITED_LIMIT) {
430-
const oldest = exitOrder[0]
431-
if (!oldest) break
432-
yield* removeCommand(oldest)
528+
// Evict per class: quick foreground commands never push out background work.
529+
const isBackground = (shellID: Shell.ID) => commands.get(shellID)?.info.metadata.background === true
530+
for (const background of [false, true]) {
531+
const cap = background ? BACKGROUND_EXITED_LIMIT : EXITED_LIMIT
532+
const members = exitOrder.filter((shellID) => isBackground(shellID) === background)
533+
for (const oldest of members.slice(0, Math.max(0, members.length - cap))) yield* removeCommand(oldest)
433534
}
434535
// Keep exited history data-only.
435536
const timeoutFiber = command.timeoutFiber
@@ -452,7 +553,18 @@ const layer = () =>
452553
return command.info
453554
})
454555

455-
return Service.of({ create, list, get, wait, result, timeout, output, remove })
556+
return Service.of({
557+
create,
558+
list,
559+
get,
560+
wait,
561+
result,
562+
timeout,
563+
background: backgroundCommand,
564+
kill,
565+
output,
566+
remove,
567+
})
456568
}),
457569
)
458570

‎packages/core/test/config/normalization.test.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,16 @@ describe("ConfigNormalize", () => {
8686
])
8787
})
8888

89+
test("keeps the shell background limit and rejects a negative one", () => {
90+
const kept = normalized({ experimental: { shell_background_after: 30_000 } })
91+
expect(kept.diagnostics).toEqual([])
92+
expect(Schema.decodeUnknownSync(Info)(kept.encoded).experimental?.shell_background_after).toBe(30_000)
93+
const zero = normalized({ experimental: { shell_background_after: 0 } })
94+
expect(Schema.decodeUnknownSync(Info)(zero.encoded).experimental?.shell_background_after).toBe(0)
95+
const negative = normalized({ experimental: { shell_background_after: -1 } })
96+
expect(negative.diagnostics.map((item) => item.path)).toEqual([["experimental", "shell_background_after"]])
97+
})
98+
8999
test("rejects every non-object root with one root diagnostic", () => {
90100
for (const input of [null, [], "config", true, 1]) {
91101
expect(ConfigNormalize.normalize(input)).toEqual({

0 commit comments

Comments
 (0)