diff --git a/packages/api/session-controller/src/client/index.ts b/packages/api/session-controller/src/client/index.ts index 4bad2bb8c2..78a54b37ef 100644 --- a/packages/api/session-controller/src/client/index.ts +++ b/packages/api/session-controller/src/client/index.ts @@ -3,6 +3,7 @@ import type { Context } from '@deepseek-ai/cordis' import type {} from '@deepseek-ai/dsh-agent/types' import { createSessionControlStream } from './transport.ts' +import { ClientJobOutput, ClientJobOutputModel } from './job-output.ts' import { ClientSessions } from './sessions/service.ts' import type { SessionRemotes } from './sessions/remotes.ts' import type {} from '../remote-events.ts' @@ -23,6 +24,8 @@ export type { } from './transport.ts' export { createScope, scopeOf } from './scope.ts' export type { AgentContext, AgentScopeHandle } from './scope.ts' +export { ClientJobOutput, ClientJobOutputModel } from './job-output.ts' +export type { IJobOutput, JobOutputSnapshot, JobOutputSource, ObservedJob } from './job-output.ts' export { SessionCreateError, SessionForkError } from './sessions/service.ts' export type { SessionBinding, SessionListState, SessionSummary } from './sessions/service.ts' export type { @@ -89,6 +92,7 @@ export const inject = [ export function apply(ctx: Context): void { const remotes = ctx.remote as unknown as SessionRemotes const sessions = new ClientSessions(ctx, remotes) + new ClientJobOutput(ctx, remotes, new ClientJobOutputModel()) ctx.remote.$on('api-session/added', (summary) => { sessions.handleSessionAdded(summary) }) ctx.remote.$on('api-session/removed', (sessionId) => { sessions.handleSessionRemoved(sessionId) }) ctx.remote.$on('api-session/status', (sessionId, running) => { diff --git a/packages/api/session-controller/src/client/job-output.ts b/packages/api/session-controller/src/client/job-output.ts new file mode 100644 index 0000000000..933c104497 --- /dev/null +++ b/packages/api/session-controller/src/client/job-output.ts @@ -0,0 +1,348 @@ +/** + * React-free client job-record observation: per-job accumulated output views + * over the `session.observeJob` stream, reference-counted per job so + * overlapping panels share one stream. The roster itself rides the session + * control stream (`jobsBySession`); this module carries only the output. + * @module @deepseek-ai/dsh-api-session-controller/client/job-output + */ + +import { Service, type Context } from '@deepseek-ai/cordis' +import { RemoteStreamCarrierError, type ClientRemote } from '@deepseek-ai/dsh-api-gateway/client' +import { notifySubscribers } from '@deepseek-ai/dsh-client-store' +import type { JobId } from '@deepseek-ai/dsh-jobs/brand' +import type { SessionId } from '@deepseek-ai/dsh-session/types' +import type { SessionJob, SessionObserveJobFrame } from '../types.ts' +import type { SessionRemote } from './transport.ts' + +/** Bounded per-job render tail, in UTF-16 code units. */ +const RENDER_TAIL_LIMIT = 128 * 1024 + +/** One observed job's live view state. */ +export interface ObservedJob { + readonly jobId: JobId + /** Accumulated output tail, bounded to the render limit. */ + readonly text: string + /** True when bytes before {@link text} were dropped (eviction, resume gap, or the render bound). */ + readonly gapBefore: boolean + /** Latest status delivered on the observation stream. */ + readonly status: SessionJob['status'] + readonly detail?: string + /** True while the observation stream is open and the job has not settled. */ + readonly streaming: boolean + /** Terminal observation failure, when the stream ended abnormally. */ + readonly error?: string +} + +/** Immutable client job-output state. */ +export interface JobOutputSnapshot { + /** Live observation state keyed by job id. */ + readonly observed: Readonly> +} + +/** Mutable observation bookkeeping behind one {@link ObservedJob} view. */ +interface ObservedState { + view: ObservedJob + /** Resume offset for the next generation's `from`. */ + cursor: number | undefined +} + +/** + * Owns per-job observation state. Pure data plus subscriptions — transport + * wiring stays in the client service, UI stays in slot components. + */ +export class ClientJobOutputModel { + private readonly observedStates = new Map() + private readonly listeners = new Set<() => void>() + private snapshotCache: JobOutputSnapshot = { observed: {} } + private snapshotDirty = false + + /** + * Read the identity-stable current snapshot. + * @returns the cached snapshot, rebuilt only after a change. + */ + getSnapshot(): JobOutputSnapshot { + if (this.snapshotDirty) { + const observed: Record = {} + for (const [id, state] of this.observedStates) observed[id] = state.view + this.snapshotCache = { observed } + this.snapshotDirty = false + } + return this.snapshotCache + } + + /** + * Subscribe to snapshot changes. + * @param listener - invalidation callback. + * @returns unsubscribe function. + */ + subscribe(listener: () => void): () => void { + this.listeners.add(listener) + return () => { this.listeners.delete(listener) } + } + + /** + * The resume offset for one job's next observation generation. + * @param id - observed job. + * @returns the last accepted `next`, or undefined for a fresh observation. + */ + cursorOf(id: JobId): number | undefined { + return this.observedStates.get(String(id))?.cursor + } + + /** + * Install or reset observation state when a generation's anchor arrives. + * @param id - observed job. + * @param frame - the generation's `opened` anchor. + */ + observeOpened(id: JobId, frame: Extract): void { + const existing = this.observedStates.get(String(id)) + // A fresh view anchored past offset zero starts after an evicted head + // (fresh observations anchor at the registry's earliest retained byte), so + // it owes the same gap mark a live observer earned from lossy reads. A + // resume that already accumulated text keeps its recorded gap state. + const freshPastHead = (existing === undefined || existing.view.text === '') && frame.from > 0 + const view: ObservedJob = { + jobId: id, + text: existing?.view.text ?? '', + gapBefore: (existing?.view.gapBefore ?? false) || frame.from < frame.earliest || freshPastHead, + status: frame.status, + ...frame.detail !== undefined ? { detail: frame.detail } : {}, + streaming: true, + } + this.observedStates.set(String(id), { view, cursor: frame.from }) + this.changed() + } + + /** + * Append one output frame's chunks to the bounded render tail. + * @param id - observed job. + * @param frame - a coalesced `output` frame. + */ + observeOutput(id: JobId, frame: Extract): void { + const state = this.observedStates.get(String(id)) + /* v8 ignore next -- frames arrive only between opened and stop for a tracked id. */ + if (state === undefined) return + let text = state.view.text + frame.chunks.map(chunk => chunk.text).join('') + let gapBefore = state.view.gapBefore || frame.lossy === true + || frame.chunks.some(chunk => chunk.gapBefore === true) + if (text.length > RENDER_TAIL_LIMIT) { + let cut = text.length - RENDER_TAIL_LIMIT + // Never split a surrogate pair at the render bound. + const unit = text.charCodeAt(cut) + if (unit >= 0xDC00 && unit <= 0xDFFF) cut += 1 + text = text.slice(cut) + gapBefore = true + } + state.view = { ...state.view, text, gapBefore } + state.cursor = frame.next + this.changed() + } + + /** + * Record the terminal status delivered at the end of the observation stream. + * @param id - observed job. + * @param frame - the terminal `status` frame. + */ + observeStatus(id: JobId, frame: Extract): void { + const state = this.observedStates.get(String(id)) + /* v8 ignore next -- frames arrive only between opened and stop for a tracked id. */ + if (state === undefined) return + state.view = { + ...state.view, + status: frame.status, + ...frame.detail !== undefined ? { detail: frame.detail } : {}, + streaming: false, + } + this.changed() + } + + /** + * Record a terminal observation failure. + * @param id - observed job. + * @param error - the stream's terminal failure. + */ + observeFailed(id: JobId, error: unknown): void { + const state = this.observedStates.get(String(id)) + if (state === undefined) return + state.view = { ...state.view, streaming: false, error: String(error) } + this.changed() + } + + /** + * Drop observation state after the last observer stops. + * @param id - the no-longer-observed job. + */ + observeStopped(id: JobId): void { + if (!this.observedStates.delete(String(id))) return + this.changed() + } + + private changed(): void { + this.snapshotDirty = true + notifySubscribers(this.listeners, 'job output') + } +} + +/** Bare observable source for the client job-output snapshot. */ +export interface JobOutputSource { + /** Read the identity-stable current snapshot. */ + getSnapshot(): JobOutputSnapshot + /** + * Subscribe to snapshot changes. + * @param listener - invalidation callback. + * @returns unsubscribe function. + */ + subscribe(listener: () => void): () => void +} + +/** The job-output client service face. */ +export interface IJobOutput { + /** Per-job observation state. */ + readonly state: JobOutputSource + /** + * Start observing one job's live record; reference-counted, so two viewers + * of the same job share one stream. + * @param sessionId - owning session used for the fenced read; undefined for an unowned job. + * @param id - job to observe. + * @returns stop function releasing this observer's reference. + */ + observe(sessionId: SessionId | undefined, id: JobId): () => void +} + +/** One reference-counted observation stream. */ +interface ObservationEntry { + refs: number + stopped: boolean + dispose: () => Promise +} + +/** Remote face the observation runner drives. */ +type JobObserveRemote = Pick & { readonly session: Pick } + +declare module '@deepseek-ai/cordis' { + interface Context { + /** React-free client job-record observation control. */ + jobOutput: IJobOutput + } +} + +/** Owns the bare job-output snapshot and per-job observation streams. */ +export class ClientJobOutput extends Service implements IJobOutput { + readonly state: JobOutputSource + private readonly entries = new Map() + + /** + * @param ctx - client root Context. + * @param remote - generated session namespace plus the Gateway stream factory. + * @param model - shared client job-output model. + */ + constructor( + ctx: Context, + private readonly remote: JobObserveRemote, + private readonly model: ClientJobOutputModel, + ) { + super(ctx, 'jobOutput') + this.state = model + ctx.effect(() => () => { + const open = [...this.entries.values()] + this.entries.clear() + for (const entry of open) { + entry.stopped = true + void entry.dispose() + } + }, 'session-controller.client.observations') + } + + observe(sessionId: SessionId | undefined, id: JobId): () => void { + const key = String(id) + const existing = this.entries.get(key) + if (existing !== undefined && !existing.stopped) { + existing.refs += 1 + return this.releaser(key, existing) + } + const entry = this.startObservation(sessionId, id) + this.entries.set(key, entry) + return this.releaser(key, entry) + } + + /** + * Release closures bind the exact entry they were minted for, never the + * map's current occupant: a later `observe()` on the same id may have + * replaced a stopped entry, and decrementing or disposing through the key + * alone would tear down that newer stream's references. + */ + private releaser(key: string, entry: ObservationEntry): () => void { + let released = false + return () => { + if (released) return + released = true + entry.refs -= 1 + if (entry.refs > 0) return + if (this.entries.get(key) === entry) this.entries.delete(key) + entry.stopped = true + void entry.dispose().then(() => { + // Clear the view only while no successor observation holds the key: + // a re-expand inside the dispose round-trip already re-anchored the + // model, and a stale clear would blank its panel for good. + if (this.entries.has(key)) return + // The key is the stringified branded id this releaser was minted for. + this.model.observeStopped(key as JobId) + }) + } + } + + private startObservation(sessionId: SessionId | undefined, id: JobId): ObservationEntry { + const name = `job observation ${String(id)}` + const stream = this.remote.$stream({ + name, + open: (signal) => { + const from = this.model.cursorOf(id) + return this.remote.session.observeJob( + { + jobId: id, + ...sessionId !== undefined ? { sessionId } : {}, + ...from !== undefined ? { from } : {}, + }, + signal, + ) + }, + // A premature end after the anchor is retryable (a Host reload closes the + // generation); resuming from the cursor loses nothing. An end before the + // anchor is terminal. + ended: accepted => accepted + ? new RemoteStreamCarrierError(`${name} ended before settlement`) + : new Error(`${name} ended before its anchor`), + }) + const entry: ObservationEntry = { + refs: 1, + stopped: false, + dispose: () => stream.dispose(), + } + void (async () => { + try { + for await (const item of stream) { + const frame = item.value + if (frame.type === 'opened') { + this.model.observeOpened(id, frame) + item.accept() + continue + } + if (frame.type === 'output') { + this.model.observeOutput(id, frame) + continue + } + // Terminal status: leave the loop before the generation end is + // classified, then close the stream for good. + this.model.observeStatus(id, frame) + break + } + } catch (error) { + if (!entry.stopped) this.model.observeFailed(id, error) + } finally { + entry.stopped = true + void entry.dispose() + } + })() + return entry + } +} diff --git a/packages/api/session-controller/src/control.ts b/packages/api/session-controller/src/control.ts index 1a03011644..aa9f0a77be 100644 --- a/packages/api/session-controller/src/control.ts +++ b/packages/api/session-controller/src/control.ts @@ -212,5 +212,6 @@ function jobView(job: JobSnapshot): SessionJob { ...(job.detail === undefined ? {} : { detail: job.detail }), startedAt: job.startedAt, ...(job.finishedAt === undefined ? {} : { finishedAt: job.finishedAt }), + ...(job.outputTotal === undefined ? {} : { outputTotal: job.outputTotal }), } } diff --git a/packages/api/session-controller/src/index.ts b/packages/api/session-controller/src/index.ts index 572f895754..99dad0a75e 100644 --- a/packages/api/session-controller/src/index.ts +++ b/packages/api/session-controller/src/index.ts @@ -2,6 +2,7 @@ import { Context } from '@deepseek-ai/cordis' import z from '@deepseek-ai/schemastery' +import type {} from '@deepseek-ai/dsh-jobs' import { errorChain } from '@deepseek-ai/dsh-llm' import { canOpenNativePath, openNativePath } from '@deepseek-ai/dsh-native-command' import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session' @@ -15,6 +16,7 @@ import { import { SessionCommandController } from './commands.ts' import { SessionControlController } from './control.ts' import { SessionHistoryController } from './history.ts' +import { observeJobRecord } from './observe-job.ts' import { SessionFileReferences } from './file-references.ts' import { ApiSessionList, DEFAULT_COLD_BLANK_PROBE_MAX_BYTES } from './list.ts' import { buildModelCatalog } from './catalog.ts' @@ -35,6 +37,8 @@ import type { SessionForkValue, SessionListRequest, SessionListValue, + SessionObserveJobFrame, + SessionObserveJobRequest, SessionOpenWorkspacePathRequest, SessionOpenWorkspacePathValue, SessionPage, @@ -63,12 +67,22 @@ declare module '@deepseek-ai/cordis' { } } +/** Default coalescing window between job-observation reads, in milliseconds. */ +const DEFAULT_OBSERVE_FLUSH_MS = 100 + +/** Default soft byte budget per job-observation output frame. */ +const DEFAULT_OBSERVE_MAX_FRAME_BYTES = 64 * 1024 + /** Session Controller deployment policy. */ export interface Config { /** Maximum cold Session artifact size eligible for one full projection observation. */ readonly coldBlankProbeMaxBytes?: number /** Override platform desktop-opener detection. */ readonly nativeOpen?: boolean + /** Coalescing window after new record output before a job-observation read, in milliseconds (default 100). */ + readonly observeFlushMs?: number + /** Soft byte budget per job-observation output frame (default 65536); one larger chunk ships whole. */ + readonly observeMaxFrameBytes?: number } /** Host integrations replaceable by direct unit tests. */ @@ -96,6 +110,8 @@ export class SessionController extends TypertRemoteService { static Config: z = z.object({ coldBlankProbeMaxBytes: z.natural().default(DEFAULT_COLD_BLANK_PROBE_MAX_BYTES), nativeOpen: z.boolean(), + observeFlushMs: z.natural().min(1).default(DEFAULT_OBSERVE_FLUSH_MS), + observeMaxFrameBytes: z.natural().min(1).default(DEFAULT_OBSERVE_MAX_FRAME_BYTES), }) private readonly agents: ApiSessionAgentController @@ -106,6 +122,8 @@ export class SessionController extends TypertRemoteService { private readonly openPath: (path: string, signal: AbortSignal) => Promise private readonly canOpenPath: () => boolean private readonly promotions = new Set>() + private readonly observeFlushMs: number + private readonly observeMaxFrameBytes: number /** * @param ctx - Host context containing the Session capability assembly. @@ -130,6 +148,8 @@ export class SessionController extends TypertRemoteService { this.openPath = internals.openPath ?? openNativePath this.canOpenPath = internals.canOpenPath ?? (() => config.nativeOpen ?? (internals.openPath !== undefined || canOpenNativePath())) + this.observeFlushMs = config.observeFlushMs ?? DEFAULT_OBSERVE_FLUSH_MS + this.observeMaxFrameBytes = config.observeMaxFrameBytes ?? DEFAULT_OBSERVE_MAX_FRAME_BYTES ctx.plugin(SessionFileReferences) ctx.plugin(SessionSkillCatalog) @@ -386,6 +406,30 @@ export class SessionController extends TypertRemoteService { return this.controlState.control(signal) } + /** + * Stream one job's retained record output from an absolute byte offset, + * then its terminal status once settled and drained. Non-consuming: the + * model-facing cursor and notice state never observe these reads. The + * request's session resolves the fenced-read caller; the registry rejects a + * job the session does not own, a job without a record, or an unknown job. + * @param request - target job, owning session, and optional resume offset. + * @param signal - cancellation owned by the Remote stream carrier. + * @returns anchor, coalesced output frames, and the terminal status. + */ + @Remote({ mode: 'stream' }) + observeJob(request: SessionObserveJobRequest, signal: AbortSignal): AsyncIterable { + const registry = this.ctx.get('jobs') + if (registry === undefined) { + throw new Error('job observation unavailable: no job registry is loaded (load @deepseek-ai/dsh-jobs-local)') + } + const caller = request.sessionId === undefined ? undefined : this.ctx.agents.get(request.sessionId) + return observeJobRecord(registry, request, { + flushMs: this.observeFlushMs, + maxFrameBytes: this.observeMaxFrameBytes, + caller, + }, signal) + } + } export { buildModelCatalog } diff --git a/packages/api/session-controller/src/observe-job.ts b/packages/api/session-controller/src/observe-job.ts new file mode 100644 index 0000000000..516f48779f --- /dev/null +++ b/packages/api/session-controller/src/observe-job.ts @@ -0,0 +1,189 @@ +/** Per-job record observation generations: anchor, coalesced output, terminal status. */ + +import type { Agent } from '@deepseek-ai/dsh-agent' +import { JobId } from '@deepseek-ai/dsh-jobs' +import type { JobRegistry, JobSnapshot } from '@deepseek-ai/dsh-jobs' +import type { SessionJobWireChunk, SessionObserveJobFrame, SessionObserveJobRequest } from './types.ts' + +/** Cadence and framing bounds for one observation generation. */ +export interface ObserveJobOptions { + /** Coalescing window after a wake before reading, in milliseconds. */ + readonly flushMs: number + /** + * Soft byte budget per `output` frame. Accumulated chunks flush once the + * budget is met; one chunk larger than the budget ships whole (the + * registry's retention cap is the hard bound on any single chunk). + */ + readonly maxFrameBytes: number + /** The live owning agent for fenced reads, resolved from the request's session. */ + readonly caller: Agent | undefined +} + +function isTerminal(status: JobSnapshot['status']): boolean { + return status !== 'running' && status !== 'stopping' +} + +/** A record job's total offset; the anchor check proved the declaration and it never leaves. */ +function recordTotal(snapshot: JobSnapshot): number { + /* v8 ignore next -- a declared record never loses its offsets. */ + return snapshot.outputTotal ?? 0 +} + +/** Wake-flag waiter: a wake between waits is never lost. */ +class OutputWaiter { + private dirty = false + private resolve: (() => void) | undefined + + wake(): void { + this.dirty = true + this.resolve?.() + } + + /** + * Resolve on the next wake, immediately when one already arrived, or on abort. + * @param signal - generation cancellation. + */ + wait(signal: AbortSignal): Promise { + if (this.dirty || signal.aborted) { + this.dirty = false + return Promise.resolve() + } + return new Promise((resolve) => { + const finish = (): void => { + signal.removeEventListener('abort', finish) + /* v8 ignore next -- one wait owns the sole installed resolver. */ + if (this.resolve === finish) this.resolve = undefined + this.dirty = false + resolve() + } + this.resolve = finish + signal.addEventListener('abort', finish, { once: true }) + }) + } +} + +function sleep(ms: number, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.resolve() + return new Promise((resolve) => { + const timer = setTimeout(done, ms) + function done(): void { + clearTimeout(timer) + signal.removeEventListener('abort', done) + resolve() + } + signal.addEventListener('abort', done, { once: true }) + }) +} + +/** + * Stream one job's retained record output from an absolute offset: one + * `opened` anchor, coalesced `output` frames as the record advances, then one + * terminal `status` after the settled job is drained, after which the + * generation closes normally. Reads are non-consuming — the model-facing + * cursor and notice state never observe them; reconnecting callers resume by + * passing the last frame's `next` as `from`. Throws for a job that declared + * no record. + * @param registry - the live job registry. + * @param request - target job and optional resume offset. + * @param options - cadence, framing bounds, and the fenced-read caller. + * @param signal - generation cancellation owned by the Remote stream carrier. + * @returns the observation frame sequence for one generation. + */ +export async function* observeJobRecord( + registry: JobRegistry, + request: SessionObserveJobRequest, + options: ObserveJobOptions, + signal: AbortSignal, +): AsyncIterable { + if (request.from !== undefined && (!Number.isSafeInteger(request.from) || request.from < 0)) { + throw new Error(`invalid observe offset: expected a non-negative safe integer, got ${JSON.stringify(request.from)}`) + } + signal.throwIfAborted() + const id = JobId(String(request.jobId)) + const waiter = new OutputWaiter() + // Subscribe before the first read so an append between the anchor read and + // the wait cannot be missed. + const unsubscribe = registry.onOutput((changed) => { + if (changed === id) waiter.wake() + }) + try { + const caller = options.caller + let snapshot = registry.get(id, caller) + if (snapshot.outputTotal === undefined || snapshot.outputEarliest === undefined) { + throw new Error(`job ${id} declared no output record`) + } + let cursor = request.from ?? snapshot.outputEarliest + yield { + type: 'opened', + jobId: id, + from: cursor, + earliest: snapshot.outputEarliest, + total: snapshot.outputTotal, + status: snapshot.status, + ...snapshot.detail !== undefined ? { detail: snapshot.detail } : {}, + } + while (!signal.aborted) { + const read = registry.readRecord(id, cursor, caller) + if (read.chunks.length > 0) { + yield* outputFrames(read.chunks, read.next, read.lossy, options.maxFrameBytes) + } + cursor = read.next + snapshot = registry.get(id, caller) + if (isTerminal(snapshot.status) && cursor >= recordTotal(snapshot)) { + yield { + type: 'status', + status: snapshot.status, + ...snapshot.detail !== undefined ? { detail: snapshot.detail } : {}, + // Settlement always stamps finishedAt; the guard only discharges the optional field type. + /* v8 ignore next */ + ...snapshot.finishedAt !== undefined ? { finishedAt: snapshot.finishedAt } : {}, + } + return + } + await waiter.wait(signal) + // Let a burst accumulate so producer chatter becomes bounded frames. + await sleep(options.flushMs, signal) + } + } finally { + unsubscribe() + } +} + +/** Split one read into frames along the soft per-frame byte budget. */ +function* outputFrames( + chunks: readonly SessionJobWireChunk[], + next: number, + lossy: boolean, + maxFrameBytes: number, +): Iterable { + let batch: SessionJobWireChunk[] = [] + let batchBytes = 0 + let flaggedLossy = lossy + for (const chunk of chunks) { + batch.push(chunk) + batchBytes += Buffer.byteLength(chunk.text, 'utf8') + if (batchBytes >= maxFrameBytes) { + const last = batch[batch.length - 1] + /* v8 ignore start -- a non-empty batch always has a last chunk; the arm only discharges noUncheckedIndexedAccess. */ + const end = last === undefined ? next : last.at + Buffer.byteLength(last.text, 'utf8') + /* v8 ignore stop */ + yield { + type: 'output', + chunks: batch, + next: end, + ...flaggedLossy ? { lossy: true as const } : {}, + } + flaggedLossy = false + batch = [] + batchBytes = 0 + } + } + if (batch.length > 0 || flaggedLossy) { + yield { + type: 'output', + chunks: batch, + next, + ...flaggedLossy ? { lossy: true as const } : {}, + } + } +} diff --git a/packages/api/session-controller/src/types.ts b/packages/api/session-controller/src/types.ts index c0b25fcf2c..cf59e29ad4 100644 --- a/packages/api/session-controller/src/types.ts +++ b/packages/api/session-controller/src/types.ts @@ -469,6 +469,12 @@ export interface SessionJob { readonly detail?: string readonly startedAt: number readonly finishedAt?: number + /** + * Total UTF-8 bytes the job's observation record holds so far. Present + * exactly when the job declared a record — the row is observable through + * `session.observeJob`. + */ + readonly outputTotal?: number } /** Complete live control baseline emitted once per control stream generation. */ @@ -493,6 +499,63 @@ export type SessionControlFrame = | { readonly type: 'jobs'; readonly sessionId: SessionId; readonly jobs: readonly SessionJob[] } | ({ readonly type: 'projection' } & SessionProjectionUpdate) +/** Opens one observation stream over a job's retained record output. */ +export interface SessionObserveJobRequest { + /** Owning session used for the fenced read; omitted for an unowned job. */ + readonly sessionId?: SessionId + readonly jobId: JobId + /** + * Absolute byte offset to resume from — a previous frame's `next`. Omitted + * starts at the oldest retained byte. + */ + readonly from?: number +} + +/** One retained record chunk on the wire. `channel` widens to string like `kind`. */ +export interface SessionJobWireChunk { + /** Absolute offset of the chunk's first byte. */ + readonly at: number + readonly text: string + readonly channel?: string + /** Bytes immediately before this chunk were lost at the producer or to retention. */ + readonly gapBefore?: true +} + +/** + * Job observation stream frames: one `opened` anchor, then coalesced `output` + * batches, then — once the job has settled and its retained record is + * drained — one terminal `status`, after which the stream closes normally. + * Status rides the same stream as output so settlement can never race a + * still-open output channel. + */ +export type SessionObserveJobFrame = + | { + readonly type: 'opened' + readonly jobId: JobId + /** Offset the first `output` frame continues from. */ + readonly from: number + /** Oldest retained byte at open time; `from < earliest` means the head is gone. */ + readonly earliest: number + /** Total bytes produced at open time. */ + readonly total: number + readonly status: SessionJob['status'] + readonly detail?: string + } + | { + readonly type: 'output' + readonly chunks: readonly SessionJobWireChunk[] + /** Offset to resume from after this frame. */ + readonly next: number + /** Bytes between the requested offset and `chunks` were already evicted. */ + readonly lossy?: true + } + | { + readonly type: 'status' + readonly status: SessionJob['status'] + readonly detail?: string + readonly finishedAt?: number + } + declare module '@deepseek-ai/cordis' { interface Events { /** diff --git a/packages/api/session-controller/tests/control-jobs.host.spec.ts b/packages/api/session-controller/tests/control-jobs.host.spec.ts index f827573d03..61a9a8156a 100644 --- a/packages/api/session-controller/tests/control-jobs.host.spec.ts +++ b/packages/api/session-controller/tests/control-jobs.host.spec.ts @@ -162,6 +162,25 @@ describe('Session control jobs updates', () => { ]) }) + it('carries outputTotal exactly for record-declaring jobs', async () => { + const { ctx, agent, control } = await harness(true) + const abort = new AbortController() + const collected = collectJobs(control.control(abort.signal), 1, abort) + ctx.jobs.start({ + kind: 'bash', + label: 'observable run', + owner: agent, + record: true, + run: (job) => { + job.append('hi') + return { cancel: () => {}, done: new Promise(() => {}) } + }, + }) + + const [frame] = await collected + expect(frame?.jobs[0]?.outputTotal).toBe(2) + }) + it('fans an unowned change out to every attached session', async () => { const { ctx, control } = await harness(true) const second = ctx.sessions.create() diff --git a/packages/api/session-controller/tests/job-output.client.spec.ts b/packages/api/session-controller/tests/job-output.client.spec.ts new file mode 100644 index 0000000000..a4ca2933dc --- /dev/null +++ b/packages/api/session-controller/tests/job-output.client.spec.ts @@ -0,0 +1,413 @@ +import { describe, expect, it, vi } from 'vitest' +import { Context } from '@deepseek-ai/cordis' +import type { JobId } from '@deepseek-ai/dsh-jobs/brand' +import type { SessionId } from '@deepseek-ai/dsh-session/types' +import { ClientJobOutput, ClientJobOutputModel } from '../src/client/job-output.ts' +import type { SessionObserveJobFrame } from '../src/types.ts' + +const ID = 'bash-1' as JobId + +function opened(model: ClientJobOutputModel, over: Partial<{ from: number; earliest: number; total: number }> = {}): void { + model.observeOpened(ID, { + type: 'opened', + jobId: ID, + from: 0, + earliest: 0, + total: 0, + status: 'running', + ...over, + }) +} + +describe('ClientJobOutputModel observation', () => { + it('accumulates output, advances the resume cursor, and settles on status', () => { + const model = new ClientJobOutputModel() + opened(model) + expect(model.cursorOf(ID)).toBe(0) + model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: 'a' }, { at: 1, text: 'b' }], next: 2 }) + model.observeOutput(ID, { type: 'output', chunks: [{ at: 2, text: 'c' }], next: 3 }) + expect(model.cursorOf(ID)).toBe(3) + const view = model.getSnapshot().observed[String(ID)] + expect(view?.text).toBe('abc') + expect(view?.streaming).toBe(true) + model.observeStatus(ID, { type: 'status', status: 'completed', detail: 'exit code: 0' }) + const settled = model.getSnapshot().observed[String(ID)] + expect(settled?.streaming).toBe(false) + expect(settled?.status).toBe('completed') + expect(settled?.detail).toBe('exit code: 0') + }) + + it('keeps snapshot identity stable between changes and notifies subscribers', () => { + const model = new ClientJobOutputModel() + let notified = 0 + const unsubscribe = model.subscribe(() => { notified += 1 }) + const before = model.getSnapshot() + expect(model.getSnapshot()).toBe(before) + opened(model) + expect(notified).toBe(1) + expect(model.getSnapshot()).not.toBe(before) + unsubscribe() + model.observeStopped(ID) + expect(notified).toBe(1) + }) + + it('marks gaps from lossy frames, gap chunks, and a resume behind the retained head', () => { + const model = new ClientJobOutputModel() + opened(model, { from: 4, earliest: 8, total: 10 }) + expect(model.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true) + + const clean = new ClientJobOutputModel() + opened(clean) + clean.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: 'x', gapBefore: true }], next: 1 }) + expect(clean.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true) + + const lossy = new ClientJobOutputModel() + opened(lossy) + lossy.observeOutput(ID, { type: 'output', chunks: [], next: 5, lossy: true }) + expect(lossy.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true) + }) + + it('marks a fresh observation anchored past the evicted head', () => { + // Fresh observations anchor at the registry's earliest retained byte, so + // from === earliest > 0 means the head was already discarded. + const fresh = new ClientJobOutputModel() + opened(fresh, { from: 60_240, earliest: 60_240, total: 321_328 }) + expect(fresh.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true) + + // A retry that accumulated no text yet earns the mark the same way. + const retried = new ClientJobOutputModel() + opened(retried) + expect(retried.getSnapshot().observed[String(ID)]?.gapBefore).toBe(false) + opened(retried, { from: 6, earliest: 6, total: 6 }) + expect(retried.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true) + }) + + it('bounds the render tail without splitting a surrogate pair', () => { + const model = new ClientJobOutputModel() + opened(model) + const emoji = '😀'.repeat((64 * 1024) + 8) + model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: emoji }], next: emoji.length * 2 }) + const view = model.getSnapshot().observed[String(ID)] + expect(view?.gapBefore).toBe(true) + expect(view!.text.length).toBeLessThanOrEqual(128 * 1024) + // The bound landed between pairs: the surviving text still round-trips. + expect(/^(?:😀)+$/u.test(view!.text)).toBe(true) + }) + + it('carries the anchor detail and trims a plain-ASCII tail without a boundary shift', () => { + const model = new ClientJobOutputModel() + model.observeOpened(ID, { + type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running', detail: 'exit soon', + }) + expect(model.getSnapshot().observed[String(ID)]?.detail).toBe('exit soon') + const long = 'x'.repeat((128 * 1024) + 5) + model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: long }], next: long.length }) + const view = model.getSnapshot().observed[String(ID)] + expect(view?.text.length).toBe(128 * 1024) + expect(view?.gapBefore).toBe(true) + }) + + it('advances the cut past a low surrogate landing exactly on the bound', () => { + const model = new ClientJobOutputModel() + opened(model) + // 'z' + one emoji + odd ASCII tail puts a low surrogate exactly at the cut index. + const text = 'z😀' + 'a'.repeat((128 * 1024) - 1) + model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text }], next: text.length }) + const view = model.getSnapshot().observed[String(ID)] + expect(view?.text.length).toBe((128 * 1024) - 1) + expect(view?.text.startsWith('a')).toBe(true) + }) + + it('preserves accumulated text across a reconnect anchor and clears on stop', () => { + const model = new ClientJobOutputModel() + opened(model) + model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: 'kept' }], next: 4 }) + opened(model, { from: 4, earliest: 0, total: 4 }) + expect(model.getSnapshot().observed[String(ID)]?.text).toBe('kept') + model.observeStopped(ID) + expect(model.getSnapshot().observed[String(ID)]).toBeUndefined() + expect(model.cursorOf(ID)).toBeUndefined() + // A second stop is inert. + model.observeStopped(ID) + }) + + it('records a terminal stream failure on the live view', () => { + const model = new ClientJobOutputModel() + opened(model) + model.observeFailed(ID, new Error('carrier gone')) + const view = model.getSnapshot().observed[String(ID)] + expect(view?.streaming).toBe(false) + expect(view?.error).toContain('carrier gone') + // A failure for an untracked id is inert. + model.observeFailed('bash-9' as JobId, new Error('ignored')) + }) +}) + +/** One scripted logical stream: frames are pushed by the test, never reopened. */ +class FakeStream { + disposed = false + /** Model a carrier whose disposal surfaces as an iterator throw. */ + throwOnDispose = false + /** Hold the dispose promise open until {@link releaseDispose}. */ + deferDispose = false + private disposeRelease: (() => void) | undefined + private disposePending: Promise | undefined + private readonly frames: SessionObserveJobFrame[] = [] + private waiter: (() => void) | undefined + private failure: Error | undefined + readonly accepts: number[] = [] + + push(frame: SessionObserveJobFrame): void { + this.frames.push(frame) + this.waiter?.() + } + + poison(error: Error): void { + this.failure = error + this.waiter?.() + } + + dispose(): Promise { + if (this.throwOnDispose && this.failure === undefined) { + this.failure = new Error('carrier tore down') + } else { + this.disposed = true + } + this.waiter?.() + if (this.deferDispose) { + // Repeat disposals (releaser plus the consumer's finally) share one + // deferred promise, so releaseDispose resumes every waiter. + this.disposePending ??= new Promise((resolve) => { this.disposeRelease = resolve }) + return this.disposePending + } + return Promise.resolve() + } + + releaseDispose(): void { + this.disposeRelease?.() + } + + async *[Symbol.asyncIterator]() { + let generation = 1 + while (!this.disposed) { + if (this.failure !== undefined) throw this.failure + const frame = this.frames.shift() + if (frame === undefined) { + await new Promise((resolve) => { this.waiter = resolve }) + continue + } + yield { + generation, + value: frame, + signal: new AbortController().signal, + accept: () => { this.accepts.push(generation) }, + } + generation = 1 + } + } +} + +function bench() { + const ctx = new Context() + const model = new ClientJobOutputModel() + const streams: { options: { open: (signal: AbortSignal) => unknown }; stream: FakeStream }[] = [] + const observeCalls: unknown[] = [] + const remote = { + $stream: (options: { open: (signal: AbortSignal) => unknown }) => { + const stream = new FakeStream() + streams.push({ options, stream }) + return stream + }, + session: { + observeJob: (request: unknown) => { + observeCalls.push(request) + return { [Symbol.asyncIterator]: async function* () { /* never yields */ } } + }, + }, + } + const output = new ClientJobOutput(ctx, remote as never, model) + return { ctx, model, output, streams, observeCalls } +} + +const tick = () => new Promise(resolve => setTimeout(resolve, 0)) + +describe('ClientJobOutput observation streams', () => { + it('feeds anchor, output, and terminal frames into the model, then closes', async () => { + const { model, output, streams } = bench() + output.observe(undefined, ID) + const { stream } = streams[0]! + stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + stream.push({ type: 'output', chunks: [{ at: 0, text: 'hi' }], next: 2 }) + await tick() + expect(model.getSnapshot().observed[String(ID)]?.text).toBe('hi') + expect(stream.accepts).toHaveLength(1) + + stream.push({ type: 'status', status: 'completed' }) + await tick() + expect(model.getSnapshot().observed[String(ID)]?.streaming).toBe(false) + expect(stream.disposed).toBe(true) + }) + + it('shares one stream across observers and stops after the last release', async () => { + const { model, output, streams } = bench() + const first = output.observe(undefined, ID) + const second = output.observe(undefined, ID) + expect(streams).toHaveLength(1) + streams[0]!.stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + await tick() + + first() + first() + expect(streams[0]!.stream.disposed).toBe(false) + second() + await tick() + expect(streams[0]!.stream.disposed).toBe(true) + expect(model.getSnapshot().observed[String(ID)]).toBeUndefined() + }) + + it('a late release of a superseded observation neither tears down nor clears its successor', async () => { + const { model, output, streams } = bench() + const firstStop = output.observe(undefined, ID) + const first = streams[0]!.stream + first.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + first.push({ type: 'status', status: 'completed' }) + await tick() + // A successor observation replaces the settled entry while the first + // observer still holds its releaser. + const secondStop = output.observe(undefined, ID) + const second = streams[1]!.stream + second.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + await tick() + + firstStop() + await tick() + expect(second.disposed).toBe(false) + expect(model.getSnapshot().observed[String(ID)]).toBeDefined() + + secondStop() + await tick() + expect(second.disposed).toBe(true) + expect(model.getSnapshot().observed[String(ID)]).toBeUndefined() + }) + + it('a re-observation inside the dispose round-trip keeps its fresh view', async () => { + const { model, output, streams } = bench() + const stop = output.observe(undefined, ID) + const first = streams[0]!.stream + first.deferDispose = true + first.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + await tick() + + stop() + // Re-expand while the previous generation's dispose is still in flight. + output.observe(undefined, ID) + const second = streams[1]!.stream + second.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + await tick() + + first.releaseDispose() + await tick() + // The stale post-dispose clear must not blank the successor's view. + expect(model.getSnapshot().observed[String(ID)]).toBeDefined() + second.push({ type: 'output', chunks: [{ at: 0, text: 'alive' }], next: 5 }) + await tick() + expect(model.getSnapshot().observed[String(ID)]?.text).toBe('alive') + }) + + it('opens a fresh stream for a re-observed job and resumes from the model cursor', async () => { + const { output, streams, observeCalls, model } = bench() + const stop = output.observe('alice' as SessionId, ID) + streams[0]!.stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + streams[0]!.stream.push({ type: 'output', chunks: [{ at: 0, text: 'abc' }], next: 3 }) + await tick() + // The generation opener reads the live cursor at open time and carries the + // fenced-read session. + streams[0]!.options.open(new AbortController().signal) + expect(observeCalls.at(-1)).toEqual({ jobId: ID, sessionId: 'alice', from: 3 }) + + stop() + await tick() + output.observe(undefined, ID) + expect(streams).toHaveLength(2) + streams[1]!.options.open(new AbortController().signal) + // Stopped observation dropped the cursor: the fresh stream starts unanchored. + expect(observeCalls.at(-1)).toEqual({ jobId: ID }) + expect(model.cursorOf(ID)).toBeUndefined() + }) + + it('classifies a premature end as retryable only after the anchor', () => { + const { output, streams } = bench() + output.observe(undefined, ID) + const options = streams[0]!.options as unknown as { ended: (accepted: boolean) => Error } + expect(options.ended(true).name).toBe('RemoteStreamCarrierError') + expect(options.ended(false).name).toBe('Error') + }) + + it('records a consumer failure on the live view', async () => { + const { model, output, streams } = bench() + output.observe(undefined, ID) + const { stream } = streams[0]! + stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + await tick() + stream.poison(new Error('frame decode broke')) + await tick() + const view = model.getSnapshot().observed[String(ID)] + expect(view?.error).toContain('frame decode broke') + expect(view?.streaming).toBe(false) + }) + + it('treats a release after service disposal as inert and stays silent for post-stop failures', async () => { + const { ctx, model, output, streams } = bench() + const stop = output.observe(undefined, ID) + const { stream } = streams[0]! + stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + await tick() + // Poison wakes the consumer, but the synchronous stop lands first, so the + // failure arrives on an already-stopped entry and stays silent. + stream.poison(new Error('after stop')) + stop() + await tick() + expect(model.getSnapshot().observed[String(ID)]).toBeUndefined() + + const second = output.observe(undefined, ID) + await ctx.fiber.dispose() + // The disposal path already removed every entry; a live releaser whose + // entry is gone is inert. + second() + }) + + it('stays silent when the carrier teardown itself throws after a stop', async () => { + const { model, output, streams } = bench() + const stop = output.observe(undefined, ID) + const { stream } = streams[0]! + stream.throwOnDispose = true + stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + await tick() + // The releaser marks the entry stopped, then its dispose surfaces as an + // iterator throw — which the consumer swallows for a stopped entry. + stop() + await tick() + expect(model.getSnapshot().observed[String(ID)]?.error).toBeUndefined() + }) + + it('records a terminal stream failure and service disposal closes live streams', async () => { + const { ctx, model, output, streams } = bench() + output.observe(undefined, ID) + const failing = streams[0]!.stream + failing.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' }) + await tick() + const iterator = failing[Symbol.asyncIterator]() + void iterator + // Simulate a terminal consumer failure by disposing the underlying stream: + // the consumer loop ends without a status frame and cleans up. + const spy = vi.spyOn(model, 'observeFailed') + await failing.dispose() + await tick() + + const other = 'bash-2' as JobId + output.observe(undefined, other) + await ctx.fiber.dispose() + expect(streams.every(entry => entry.stream.disposed)).toBe(true) + expect(spy).not.toHaveBeenCalled() + }) +}) diff --git a/packages/api/session-controller/tests/observe-job.host.spec.ts b/packages/api/session-controller/tests/observe-job.host.spec.ts new file mode 100644 index 0000000000..b61079d2b1 --- /dev/null +++ b/packages/api/session-controller/tests/observe-job.host.spec.ts @@ -0,0 +1,298 @@ +import { describe, expect, it } from 'vitest' +import { Context } from '@deepseek-ai/cordis' +import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent' +import type { Agent } from '@deepseek-ai/dsh-agent' +import { JobId } from '@deepseek-ai/dsh-jobs' +import type { JobOutcome, RunningJob } from '@deepseek-ai/dsh-jobs' +import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local' +import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session' +import { observeJobRecord } from '../src/observe-job.ts' +import type { ObserveJobOptions } from '../src/observe-job.ts' +import type { SessionObserveJobFrame } from '../src/types.ts' +import { createSessionTestController } from './test-remote.ts' + +async function harness() { + const ctx = new Context() + await ctx.plugin(AgentRegistry) + await ctx.plugin(LocalJobRegistry, { retainBytes: 64, settledRetainBytes: 64 }) + ctx.jobs.attachController('observe-job-test') + return ctx +} + +function registerAgent(ctx: Context, rawId: string): Agent { + const scopeFiber = ctx.plugin(() => {}) + const id = SessionId(rawId) + const session = Session.create(id) + const agent = { + id, + session, + inbox: new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} }), + status: 'idle', + ctx: scopeFiber.ctx, + } as unknown as Agent + ctx.agents.register(agent) + return agent +} + +/** Start one record job whose done settles on demand; expose the producer face. */ +function startRecordJob(ctx: Context, options: { label?: string; owner?: Agent; record?: boolean } = {}) { + let settle!: (outcome: JobOutcome) => void + let face!: RunningJob + const id = ctx.jobs.start({ + kind: 'bash', + label: options.label ?? 'echo', + ...options.owner !== undefined ? { owner: options.owner } : {}, + ...options.record === false ? {} : { record: true }, + run: (job) => { + face = job + return { + cancel() {}, + done: new Promise((resolve) => { settle = resolve }), + } + }, + }) + return { + id, + append: (text: string, opts?: Parameters[1]) => { face.append(text, opts) }, + settle: async (outcome: JobOutcome) => { + settle(outcome) + await new Promise(resolve => setTimeout(resolve, 0)) + }, + } +} + +/** Read frames until `count` arrive or the generator finishes, then abort. */ +async function collect( + iterable: AsyncIterable, + count: number, + abort: AbortController, +): Promise { + const frames: Frame[] = [] + const iterator = iterable[Symbol.asyncIterator]() + try { + while (frames.length < count) { + const step = await iterator.next() + if (step.done) return frames + frames.push(step.value) + } + } finally { + abort.abort() + await iterator.next().catch(() => undefined) + } + return frames +} + +const OBSERVE: Omit = { flushMs: 5, maxFrameBytes: 1024 } + +describe('observeJobRecord', () => { + function observed(ctx: Context, request: { jobId: JobId; from?: number }, caller?: Agent) { + const abort = new AbortController() + const frames: SessionObserveJobFrame[] = [] + const stream = observeJobRecord(ctx.jobs, request, { ...OBSERVE, caller }, abort.signal) + const done = (async () => { + for await (const frame of stream) frames.push(frame) + })() + return { abort, frames, done } + } + + it('anchors, streams coalesced output, and closes with the terminal status', async () => { + const ctx = await harness() + const job = startRecordJob(ctx) + job.append('early ') + const { frames, done } = observed(ctx, { jobId: job.id }) + await new Promise(resolve => setTimeout(resolve, 20)) + job.append('live', { channel: 'stderr' }) + await new Promise(resolve => setTimeout(resolve, 20)) + await job.settle({ status: 'completed', detail: 'exit code: 0' }) + await done + + expect(frames[0]).toMatchObject({ type: 'opened', from: 0, earliest: 0, total: 6, status: 'running' }) + const outputs = frames.filter(frame => frame.type === 'output') + const text = outputs.flatMap(frame => frame.chunks.map(chunk => chunk.text)).join('') + expect(text).toBe('early live') + expect(outputs.at(-1)?.next).toBe(10) + expect(outputs.flatMap(frame => frame.chunks).find(chunk => chunk.text === 'live')?.channel).toBe('stderr') + expect(frames.at(-1)).toMatchObject({ type: 'status', status: 'completed', detail: 'exit code: 0' }) + }) + + it('resumes from an explicit offset and flags an evicted resume lossy', async () => { + const ctx = await harness() + const job = startRecordJob(ctx, { label: 'spam' }) + job.append('a'.repeat(60)) + job.append('b'.repeat(60)) + await job.settle({ status: 'completed' }) + + const fresh = observed(ctx, { jobId: job.id, from: 0 }) + await fresh.done + const output = fresh.frames.find(frame => frame.type === 'output') + expect(output?.lossy).toBe(true) + expect(output?.chunks.map(chunk => chunk.text).join('')).toBe('b'.repeat(60)) + + const resumed = observed(ctx, { jobId: job.id, from: 120 }) + await resumed.done + expect(resumed.frames.some(frame => frame.type === 'output')).toBe(false) + expect(resumed.frames.at(-1)?.type).toBe('status') + }) + + it('splits oversized reads along the frame budget', async () => { + const ctx = await harness() + const job = startRecordJob(ctx, { label: 'wide' }) + job.append('x'.repeat(30)) + job.append('y'.repeat(30)) + await job.settle({ status: 'completed' }) + const abort = new AbortController() + const frames = await collect( + observeJobRecord(ctx.jobs, { jobId: job.id }, { + flushMs: 5, + maxFrameBytes: 30, + caller: undefined, + }, abort.signal), + 5, + abort, + ) + const outputs = frames.filter(frame => frame.type === 'output') + expect(outputs).toHaveLength(2) + expect(outputs[0]?.next).toBe(30) + expect(outputs[1]?.next).toBe(60) + }) + + it('carries the lossy flag on the first frame of a split evicted read', async () => { + const ctx = await harness() + const job = startRecordJob(ctx, { label: 'evicted split' }) + job.append('a'.repeat(40)) + job.append('b'.repeat(40)) + job.append('c'.repeat(40)) + await job.settle({ status: 'completed' }) + const abort = new AbortController() + const frames = await collect( + observeJobRecord(ctx.jobs, { jobId: job.id, from: 0 }, { + flushMs: 5, + maxFrameBytes: 40, + caller: undefined, + }, abort.signal), + 5, + abort, + ) + const outputs = frames.filter(frame => frame.type === 'output') + expect(outputs[0]?.lossy).toBe(true) + expect(outputs[1]?.lossy).toBeUndefined() + }) + + it('reads owned records through the resolved caller and rejects foreign, bad, and unknown input', async () => { + const ctx = await harness() + const owner = registerAgent(ctx, 'alice') + const job = startRecordJob(ctx, { label: 'secret', owner }) + job.append('classified') + await job.settle({ status: 'completed' }) + + const { frames, done } = observed(ctx, { jobId: job.id }, owner) + await done + expect(frames.some(frame => frame.type === 'output' + && frame.chunks.some(chunk => chunk.text === 'classified'))).toBe(true) + + const foreign = new AbortController() + await expect(collect( + observeJobRecord(ctx.jobs, { jobId: job.id }, { ...OBSERVE, caller: undefined }, foreign.signal), + 1, + foreign, + )).rejects.toThrow(/belongs to another session/) + + const abort = new AbortController() + await expect(collect( + observeJobRecord(ctx.jobs, { jobId: job.id, from: -1 }, { ...OBSERVE, caller: owner }, abort.signal), + 1, + abort, + )).rejects.toThrow(/invalid observe offset/) + + const abortUnknown = new AbortController() + await expect(collect( + observeJobRecord(ctx.jobs, { jobId: JobId('bash-999') }, { ...OBSERVE, caller: undefined }, abortUnknown.signal), + 1, + abortUnknown, + )).rejects.toThrow(/unknown job/) + }) + + it('refuses a job that declared no record', async () => { + const ctx = await harness() + const job = startRecordJob(ctx, { record: false }) + const abort = new AbortController() + await expect(collect( + observeJobRecord(ctx.jobs, { jobId: job.id }, { ...OBSERVE, caller: undefined }, abort.signal), + 1, + abort, + )).rejects.toThrow(/declared no output record/) + await job.settle({ status: 'completed' }) + }) + + it('ignores output signals for other jobs while waiting', async () => { + const ctx = await harness() + const watched = startRecordJob(ctx, { label: 'watched' }) + const noisy = startRecordJob(ctx, { label: 'noisy' }) + const { abort, frames, done } = observed(ctx, { jobId: watched.id }) + await new Promise(resolve => setTimeout(resolve, 15)) + noisy.append('unrelated') + await new Promise(resolve => setTimeout(resolve, 15)) + await watched.settle({ status: 'completed' }) + await done + abort.abort() + expect(frames.filter(frame => frame.type === 'output')).toEqual([]) + await noisy.settle({ status: 'completed' }) + }) + + it('stops cleanly on abort while waiting for output', async () => { + const ctx = await harness() + const job = startRecordJob(ctx, { label: 'idle' }) + const { abort, frames, done } = observed(ctx, { jobId: job.id }) + await new Promise(resolve => setTimeout(resolve, 20)) + abort.abort() + await done + expect(frames[0]?.type).toBe('opened') + expect(frames.at(-1)?.type).not.toBe('status') + await job.settle({ status: 'completed' }) + }) +}) + +describe('SessionController.observeJob', () => { + const defaults = { + defaultModelSelection: () => ({ provider: 'fixture', model: 'fixture-model' }), + cwd: '/tmp', + } + + it('refuses observation without a job registry', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(AgentRegistry) + const controller = createSessionTestController(ctx, defaults) + expect(() => controller.observeJob({ jobId: JobId('bash-1') }, new AbortController().signal)) + .toThrow(/job observation unavailable/) + }) + + it('resolves the fenced-read caller from the request session', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(AgentRegistry) + await ctx.plugin(LocalJobRegistry) + ctx.jobs.attachController('observe-job-test') + const controller = createSessionTestController(ctx, defaults) + const session = ctx.sessions.create(SessionId('observing-session')) + const owner = { + id: session.id, + session, + inbox: new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} }), + status: 'idle', + ctx, + } as unknown as Agent + ctx.agents.register(owner) + const job = startRecordJob(ctx, { label: 'owned run', owner }) + job.append('hi') + await job.settle({ status: 'completed' }) + + const abort = new AbortController() + const frames = await collect( + controller.observeJob({ sessionId: session.id, jobId: job.id }, abort.signal), + 3, + abort, + ) + expect(frames.map(frame => frame.type)).toEqual(['opened', 'output', 'status']) + }) +}) diff --git a/packages/api/session-controller/tests/test-remote.ts b/packages/api/session-controller/tests/test-remote.ts index 3d7d549a26..d125b6808d 100644 --- a/packages/api/session-controller/tests/test-remote.ts +++ b/packages/api/session-controller/tests/test-remote.ts @@ -34,6 +34,8 @@ import type { SessionFollowRequest, SessionListRequest, SessionListValue, + SessionObserveJobFrame, + SessionObserveJobRequest, SessionOpenWorkspacePathRequest, SessionOpenWorkspacePathValue, SessionPage, @@ -71,6 +73,7 @@ export interface TestSessionRemote { page(request: SessionPageRequest, signal?: AbortSignal): Promise> follow(request: SessionFollowRequest, signal?: AbortSignal): AsyncIterable control(signal?: AbortSignal): AsyncIterable + observeJob(request: SessionObserveJobRequest, signal?: AbortSignal): AsyncIterable } /** Dependencies and policy supplied by a Session Controller unit harness. */ @@ -273,5 +276,6 @@ export function createSessionTestRemote( ), follow: (request, signal = new AbortController().signal) => direct.follow(request, signal), control: (signal = new AbortController().signal) => direct.control(signal), + observeJob: (request, signal = new AbortController().signal) => direct.observeJob(request, signal), } } diff --git a/packages/api/session-controller/tsconfig.host.json b/packages/api/session-controller/tsconfig.host.json index a48fcb83a5..207daafbd5 100644 --- a/packages/api/session-controller/tsconfig.host.json +++ b/packages/api/session-controller/tsconfig.host.json @@ -18,6 +18,7 @@ "src/history.ts", "src/list.ts", "src/model-selection-projection.ts", + "src/observe-job.ts", "src/skill-catalog.ts" ], "references": [