Merge remote-tracking branch 'origin/master' into codex/basic-session-search
Conflict resolutions: - `session.list`: master's projection columns fold into the PR's cancellable, batched `listVisibleSessionSummaries`, which `session.search` shares as its visibility baseline; master's goal helpers stay beside it. - Client sessions face: master narrowed `ctx.sessions` to `ISessions`, so the search verb and its protocol-constant bound are declared there and the test-runtime double implements them (recorded, empty page unless a scenario stubs hits). - `WorkspaceBrowser`: master's per-row Rename wiring rides the PR's search results view; the tree keeps the PR's query-free derivations. - `dsh web` bin: the PR's shutdown-handlers-before-readiness order with master's boot-time LAN address snapshot. - `session-query-sqlite`: master's `SCHEMA_VERSION` 7 stands; the PR's bump carried no schema change. - Specs: master wraps assistant/steering message payloads and requires an `application/json` carrier request, so the search fixtures and tests follow. - Web aria goldens keep master's recording plus the PR's search placeholder; the navigation-panes inventory keeps master's terminal-card golden next to the PR's search-results golden.
This commit is contained in:
Hypatia May
@@ -9,15 +9,14 @@ import { join } from 'node:path'
|
||||
import type { Context } from 'cordis'
|
||||
import { installAgentLlmTarget } from '@deepseek-ai/dsh-agent'
|
||||
import type {
|
||||
Agent, AgentLlmTarget, AgentLlmTargetRef, AgentMessage, AgentMessageId, AgentStatus,
|
||||
Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus, InboxPlacement,
|
||||
} from '@deepseek-ai/dsh-agent'
|
||||
import { ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
import { createUserMessage, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
import { errorChain } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import type { JsonValue, Session, SessionEvent, SessionHeader, SessionId, TodoItem } from '@deepseek-ai/dsh-session'
|
||||
import type { MessageId, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import type { Session, SessionEvent, SessionHeader, SessionId, UserMessage } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { SessionQueryError, type SessionSearchCursor } from '@deepseek-ai/dsh-session-query'
|
||||
import { foldSessionTitle } from '@deepseek-ai/dsh-session-title'
|
||||
import type { Workspace, WorkspaceRecord } from '@deepseek-ai/dsh-workspace'
|
||||
import {
|
||||
workspaceDomainState, workspaceRecord, WorkspaceId as brandWorkspaceId,
|
||||
@@ -26,18 +25,33 @@ import {
|
||||
// Type-only: brings the `ctx.tools` Context merge into this program (viewFor reads presenters).
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import type {
|
||||
ApiProxy, HistoryEntry, HostFrame, ModelCatalogFailure, ModelProviderGroup, ModelReasoning,
|
||||
MuxFrame, QuestionResponsePayload, SessionSearchItem, SessionSummary, ToolEventView,
|
||||
WorkspaceId, WorkspaceView,
|
||||
ApiProxy, GoalRef, HistoryEntry, HostFrame, ModelCatalogFailure, ModelProviderGroup, ModelReasoning,
|
||||
MuxFrame, QuestionResponsePayload, SessionProjectionsBlock, SessionSearchItem, SessionSummary,
|
||||
ToolEventView, WorkspaceId, WorkspaceView,
|
||||
} from './api/index.ts'
|
||||
import {
|
||||
SESSION_SEARCH_RESULT_LIMIT,
|
||||
SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS,
|
||||
truncateUnicodeCodePoints,
|
||||
} from './api/session-search.ts'
|
||||
// Type-only: resolves `ctx.get('sessionProjections')` to the projection registry.
|
||||
import type {} from '@deepseek-ai/dsh-session-projection'
|
||||
// Type-only: resolves `ctx.get('sessionProjectionCache')` (the cold listing column).
|
||||
import type {} from '@deepseek-ai/dsh-session-projection-cache'
|
||||
// GoalError narrows domain rejections to their stable codes at the wire boundary.
|
||||
import { GoalError } from '@deepseek-ai/dsh-goal'
|
||||
import type { GoalRef as CoreGoalRef } from '@deepseek-ai/dsh-goal'
|
||||
// Type-only edges: resolve `ctx.get('commands')`, the `commands/change` event, and `ctx.get('skills')`.
|
||||
import type {} from '@deepseek-ai/dsh-commands'
|
||||
import type {} from '@deepseek-ai/dsh-skill'
|
||||
// Value edge: the rename impl narrows the title service's validation failure; the import also resolves `ctx.get('sessionTitle')`.
|
||||
import { SessionTitleInvalidError } from '@deepseek-ai/dsh-session-title'
|
||||
import type { CallId } from '@deepseek-ai/dsh-llm/brand'
|
||||
import type { ApprovalOutcome, ApprovalRequestId } from '@deepseek-ai/dsh-user-approval'
|
||||
// Side-effect type import: resolves the `approval/request` waterfall and
|
||||
// `ctx.get('approval')` without a value dependency on the seam (optional composition).
|
||||
import type {} from '@deepseek-ai/dsh-user-approval'
|
||||
import { approvalResponsePayloadSchema } from './api/approvals.schema.ts'
|
||||
import { questionResponsePayloadSchema } from './api/questions.schema.ts'
|
||||
import type { ClientResponse, RpcError, RpcReceipt, RpcRequest, RpcResponse } from './api/rpc.ts'
|
||||
import { RpcId } from './api/rpc.ts'
|
||||
@@ -45,7 +59,8 @@ import type {
|
||||
AskUserQuestionAnswer, AskUserQuestionItem, AskUserQuestionRequest,
|
||||
} from '@deepseek-ai/dsh-user-interaction'
|
||||
import { UserInteractionError } from '@deepseek-ai/dsh-user-interaction'
|
||||
import { pickNativeDirectory } from './native-directory-picker.ts'
|
||||
import { DirectoryPickerError } from '@deepseek-ai/dsh-host-directory-picker'
|
||||
import { openNativePath } from './native-path-opener.ts'
|
||||
|
||||
/** Page size when history is called without maxMessages. */
|
||||
const DEFAULT_MAX_MESSAGES = 50
|
||||
@@ -138,34 +153,28 @@ class FrameQueue<F> {
|
||||
}
|
||||
|
||||
/**
|
||||
* Server-side frame mint: pure pushes get a fresh rpcId per frame (stable ids
|
||||
* for answerable frames belong to the approval/question registry, absent in
|
||||
* this minimal version).
|
||||
* Server-side frame mint: pure pushes get a fresh rpcId per frame (answerable
|
||||
* frames — approval/question requested — mint their stable id in their
|
||||
* pending registries instead).
|
||||
*/
|
||||
function frame<F>(payload: F): RpcRequest<F> {
|
||||
return { rpcId: RpcId(randomUUID()), payload }
|
||||
}
|
||||
|
||||
type SessionTitleFrame = Extract<MuxFrame, { type: 'session/title' }>
|
||||
|
||||
/** Project the latest durable title without exposing title-generation policy. */
|
||||
function titleFrame(session: Session): SessionTitleFrame | undefined {
|
||||
const title = foldSessionTitle(session.events)
|
||||
if (title === undefined) return undefined
|
||||
return {
|
||||
type: 'session/title',
|
||||
sessionId: session.id,
|
||||
title: title.title,
|
||||
eventSeq: title.eventSeq,
|
||||
updatedAt: title.updatedAt,
|
||||
}
|
||||
}
|
||||
|
||||
/** Queue the subscription baseline followed by its optional title snapshot. */
|
||||
/** Queue the subscription baseline frame. */
|
||||
function subscribeSession(queue: FrameQueue<RpcRequest<MuxFrame>>, session: Session): void {
|
||||
queue.push(frame({ type: 'session/subscribed', sessionId: session.id, lastSeq: session.seq - 1 }))
|
||||
const title = titleFrame(session)
|
||||
if (title !== undefined) queue.push(frame(title))
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether the session's conversation has started: no turn has run yet (a
|
||||
* turn is one model-loop execution). Standalone plugin events — command
|
||||
* lifecycle records, plan/mode, titles, goals — never open a turn, so
|
||||
* running `/plan` or `/goal` on a fresh session keeps it blank
|
||||
* (list-hidden, reusable).
|
||||
*/
|
||||
function sessionBlank(session: Session): boolean {
|
||||
return !session.events.some(event => event.type === 'turn/start')
|
||||
}
|
||||
|
||||
/** SessionSummary projection for attached (in-memory) sessions. */
|
||||
@@ -174,7 +183,7 @@ function summarize(session: Session, running: boolean): SessionSummary {
|
||||
sessionId: session.id,
|
||||
updatedAt: session.events.at(-1)?.time ?? session.header.createdAt,
|
||||
running,
|
||||
blank: session.events.length === 0,
|
||||
blank: sessionBlank(session),
|
||||
...session.header.parentSession === undefined ? {} : { parentSessionId: session.header.parentSession },
|
||||
...session.header.cwd === undefined ? {} : { cwd: session.header.cwd },
|
||||
}
|
||||
@@ -206,8 +215,9 @@ async function summarizeCold(
|
||||
sessionId: meta.id,
|
||||
updatedAt,
|
||||
running: false,
|
||||
// Lazy persistence keeps never-appended sessions out of list(): a cold
|
||||
// session necessarily has events, so blank is constantly false here.
|
||||
// Lazy persistence keeps never-appended sessions out of list(); reading
|
||||
// a cold log to check for turns would defeat the index read, so a listed
|
||||
// cold session is served as not-blank (its log holds its conversation).
|
||||
blank: false,
|
||||
...meta.parentSession === undefined ? {} : { parentSessionId: meta.parentSession },
|
||||
/* v8 ignore next -- the empty arm needs a cwd-less meta, but list()
|
||||
@@ -217,6 +227,14 @@ async function summarizeCold(
|
||||
}
|
||||
}
|
||||
|
||||
/** Map a browse-primitive failure onto the wire error vocabulary (unknown throws stay internal). */
|
||||
function directoryError(error: unknown): RpcError {
|
||||
if (error instanceof DirectoryPickerError) {
|
||||
return { code: error.code, message: error.message, details: { path: error.path } }
|
||||
}
|
||||
return { code: 'internal', message: error instanceof Error ? error.message : String(error), details: {} }
|
||||
}
|
||||
|
||||
/** Resolved Host routing and project-directory defaults consumed by the API implementation. */
|
||||
export interface ApiProxyDefaults {
|
||||
provider: string
|
||||
@@ -225,14 +243,41 @@ export interface ApiProxyDefaults {
|
||||
cwd: string
|
||||
/** Parent directory for name-created workspaces. */
|
||||
workspaceRoot: string
|
||||
/** Native single-directory picker; injectable for carrier tests. */
|
||||
pickDirectory?: (signal: AbortSignal) => Promise<string | null>
|
||||
/** Native open-with-default-application; injectable for carrier tests. */
|
||||
openPath?: (path: string, signal: AbortSignal) => Promise<void>
|
||||
}
|
||||
|
||||
/** The tool/call payload fields the presenter path reads. */
|
||||
interface ToolCallData { callId: string; name: string; arguments: string }
|
||||
/** The tool/result payload fields the presenter path reads. */
|
||||
interface ToolResultData { callId: string; content: ContentBlock[]; isError: boolean; meta?: JsonValue }
|
||||
/**
|
||||
* One outstanding approval question: the stable server-request id, the frame
|
||||
* material replayed to late mux subscribers, and the resolver that settles the
|
||||
* answerer's promise back into `ctx.approval`.
|
||||
*/
|
||||
interface PendingApproval {
|
||||
rpcId: RpcId
|
||||
sessionId: SessionId
|
||||
approvalId: ApprovalRequestId
|
||||
toolName: string
|
||||
callId?: CallId
|
||||
reason?: string
|
||||
resolve(outcome: ApprovalOutcome): void
|
||||
}
|
||||
|
||||
/** Project a pending entry into its answerable mux frame (initial push and mux-open replay share it). */
|
||||
function requestedFrame(pending: PendingApproval): RpcRequest<MuxFrame> {
|
||||
return {
|
||||
rpcId: pending.rpcId,
|
||||
payload: {
|
||||
type: 'approval/requested',
|
||||
sessionId: pending.sessionId,
|
||||
approvalId: pending.approvalId,
|
||||
toolName: pending.toolName,
|
||||
...pending.callId === undefined ? {} : { callId: pending.callId },
|
||||
...pending.reason === undefined ? {} : { reason: pending.reason },
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/** One host-owned question wait, addressed by the stable server-request id. */
|
||||
interface PendingQuestion {
|
||||
@@ -280,10 +325,16 @@ function viewFor(ctx: Context, event: SessionEvent, argsFor: (callId: string) =>
|
||||
return view === undefined ? undefined : { for: 'call', view }
|
||||
}
|
||||
if (event.type === 'tool/result') {
|
||||
const { callId, content, isError, meta } = event.data as ToolResultData
|
||||
const { message, meta } = event.data
|
||||
const [result] = message.content
|
||||
const callId = message.source.callId
|
||||
const call = argsFor(callId) as { name: string; args: unknown } | undefined
|
||||
if (call === undefined) return undefined
|
||||
const view = ctx.tools.get(call.name)?.presentResult?.(call.args, { content, isError, ...meta === undefined ? {} : { meta } })
|
||||
const view = ctx.tools.get(call.name)?.presentResult?.(call.args, {
|
||||
content: result.content,
|
||||
isError: result.isError === true,
|
||||
...meta === undefined ? {} : { meta },
|
||||
})
|
||||
return view === undefined ? undefined : { for: 'result', view }
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
@@ -316,13 +367,41 @@ function backscanArgs(events: readonly SessionEvent[], callId: string): { name:
|
||||
return undefined
|
||||
}
|
||||
|
||||
/** Current todo projection: the latest `todo/write` over the full log (whole-list replace ⇒ last write wins); undefined when none. */
|
||||
function backscanTodos(events: readonly SessionEvent[]): TodoItem[] | undefined {
|
||||
for (let i = events.length - 1; i >= 0; i--) {
|
||||
const event = events[i]
|
||||
if (event !== undefined && event.type === 'todo/write') return event.data.todos
|
||||
/**
|
||||
* The projection baseline for one history tail page: the registry's
|
||||
* watermark-cache snapshot — one fully synchronous read (no await between the
|
||||
* page slice and this), so all values and `asOfSeq` form a single consistent
|
||||
* cut and `asOfSeq` equals the window tail event seq. The carrier holds zero
|
||||
* domain knowledge (each value passed its unit's own schema inside the
|
||||
* registry). An absent registry means the deployment has no projection seam:
|
||||
* the whole block is absent and clients treat every key as capability-absent.
|
||||
*/
|
||||
function projectionsFor(ctx: Context, agent: Agent): SessionProjectionsBlock | undefined {
|
||||
const registry = ctx.get('sessionProjections')
|
||||
if (registry === undefined) return undefined
|
||||
return registry.snapshot(agent.session)
|
||||
}
|
||||
|
||||
/**
|
||||
* The projection baseline of one session.list row, fail-soft: attached
|
||||
* sessions cut the registry's live watermark cache; cold sessions view the
|
||||
* persisted projection cache's identity-checked stored rows (zero log loads
|
||||
* either way — the listing use case the cache exists for). The block shape
|
||||
* (values + asOfSeq) matches the history tail's, so a client seeds its
|
||||
* value store under the same higher-seq-wins rule. Any failure — and an
|
||||
* empty value set — yields an absent block: a listing without projections
|
||||
* is degraded, never broken.
|
||||
*/
|
||||
function listProjectionsFor(ctx: Context, meta: SessionHeader, session: Session | undefined): SessionProjectionsBlock | undefined {
|
||||
try {
|
||||
const block = session !== undefined
|
||||
? ctx.get('sessionProjections')?.snapshot(session)
|
||||
: ctx.get('sessionProjectionCache')?.cachedSnapshot(meta)
|
||||
return block !== undefined && Object.keys(block.values).length > 0 ? block : undefined
|
||||
} catch (error) {
|
||||
ctx.logger.warn(`session.list: projection column for "${meta.id}" failed (serving the row without it): ${String(error)}`)
|
||||
return undefined
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -399,6 +478,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
/** Serializes path ownership checks with record creation across spellings. */
|
||||
let workspaceCreationChain = Promise.resolve()
|
||||
const pendingQuestions = new Map<RpcId, PendingQuestion>()
|
||||
const pendingApprovals = new Map<RpcId, PendingApproval>()
|
||||
const muxQueues = new Set<FrameQueue<RpcRequest<MuxFrame>>>()
|
||||
|
||||
/**
|
||||
@@ -441,42 +521,53 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
for (const queue of muxQueues) queue.push(envelope)
|
||||
}
|
||||
|
||||
// Projection change feed → session/projection push frames. The carrier
|
||||
// mints the wire frame (the seam package holds no wire vocabulary); the
|
||||
// child activates only when a projection registry is composed, and the
|
||||
// subscription unwinds with this gateway's fiber.
|
||||
ctx.inject(['sessionProjections'], (projectionCtx) => {
|
||||
projectionCtx.sessionProjections.onChanged((session, key, value, seq) => {
|
||||
broadcast({ type: 'session/projection', sessionId: session.id, key, value, seq })
|
||||
})
|
||||
})
|
||||
|
||||
/**
|
||||
* Per-session inbox mirror serving the mux-open queue snapshot (the same
|
||||
* refresh-recovery baseline as pending questions). Keyed by the stable
|
||||
* AgentMessageId: every enqueued id receives exactly one terminal
|
||||
* `agent/inbox/dequeue` OR `agent/inbox/discard` (the inbox contract), so
|
||||
* the mirror needs no consumption heuristics or sweeps beyond disposal.
|
||||
* Per-session inbox occurrence mirror serving the mux-open queue snapshot
|
||||
* (the same refresh-recovery baseline as pending questions). Each terminal
|
||||
* inbox event retires one matching occurrence, so repeated sends of the same
|
||||
* identified message remain visible until every occurrence is claimed.
|
||||
*/
|
||||
const queuedMirror = new Map<SessionId, Map<AgentMessageId, { message: AgentMessage; steering: boolean }>>()
|
||||
const queuedMirror = new Map<SessionId, { message: UserMessage; steering: boolean }[]>()
|
||||
ctx.effect(() => {
|
||||
const retire = (agent: Agent, id: AgentMessageId): void => {
|
||||
const retire = (agent: Agent, id: MessageId, placement?: InboxPlacement): void => {
|
||||
const entries = queuedMirror.get(agent.id)
|
||||
if (entries === undefined) return
|
||||
entries.delete(id)
|
||||
if (entries.size === 0) queuedMirror.delete(agent.id)
|
||||
const index = entries.findIndex(entry =>
|
||||
entry.message.id === id
|
||||
&& (placement === undefined || entry.steering === (placement === 'steering')))
|
||||
if (index !== -1) entries.splice(index, 1)
|
||||
if (entries.length === 0) queuedMirror.delete(agent.id)
|
||||
}
|
||||
const disposers = [
|
||||
ctx.on('agent/inbox/enqueue', (agent: Agent, message: AgentMessage, placement) => {
|
||||
ctx.on('agent/inbox/enqueue', (agent: Agent, message: UserMessage, placement) => {
|
||||
let entries = queuedMirror.get(agent.id)
|
||||
if (entries === undefined) {
|
||||
entries = new Map<AgentMessageId, { message: AgentMessage; steering: boolean }>()
|
||||
entries = []
|
||||
queuedMirror.set(agent.id, entries)
|
||||
}
|
||||
const steering = placement === 'steering'
|
||||
entries.set(message.id, { message, steering })
|
||||
entries.push({ message, steering })
|
||||
broadcast({
|
||||
type: 'session/queued',
|
||||
sessionId: agent.id,
|
||||
content: message.content,
|
||||
source: message.source,
|
||||
message,
|
||||
steering,
|
||||
})
|
||||
}),
|
||||
ctx.on('agent/inbox/dequeue', (agent: Agent, message: AgentMessage) => {
|
||||
retire(agent, message.id)
|
||||
ctx.on('agent/inbox/dequeue', (agent: Agent, message: UserMessage, placement) => {
|
||||
retire(agent, message.id, placement)
|
||||
}),
|
||||
ctx.on('agent/inbox/discard', (agent: Agent, messages: AgentMessage[]) => {
|
||||
ctx.on('agent/inbox/discard', (agent: Agent, messages: UserMessage[]) => {
|
||||
for (const message of messages) retire(agent, message.id)
|
||||
}),
|
||||
ctx.on('session/disposed', (session: Session) => {
|
||||
@@ -536,6 +627,90 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
}
|
||||
}, 'api-proxy: user-interaction provider')
|
||||
|
||||
// --- Approval pending registry ------------------------------------------
|
||||
// The proxy is the approval channel for every agent this host owns: an ask
|
||||
// through `ctx.approval` becomes an answerable server-request on the mux
|
||||
// stream (stable rpcId), settled by POST /api/respond. The entry survives
|
||||
// client disconnects — mux-open replays still-pending requested frames with
|
||||
// the same rpcId (the refresh-recovery baseline) — and withdraws on the
|
||||
// ask's own abort signal (turn cancel), pushing `cancelled` to subscribers.
|
||||
if (ctx.get('approval') !== undefined) {
|
||||
// Teardown parity with the question provider above: a gateway disposed
|
||||
// while approvals are pending settles every entry as 'cancelled' (the
|
||||
// service's fail-closed vocabulary), so no ask promise dangles past the
|
||||
// proxy's lifetime and subscribers see the withdrawal.
|
||||
ctx.effect(() => () => {
|
||||
for (const pending of [...pendingApprovals.values()]) pending.resolve('cancelled')
|
||||
}, 'api-proxy: approval registry teardown')
|
||||
ctx.on('approval/request', (req, next) => {
|
||||
// Dispatch rides a microtask behind the service's own signal check: an
|
||||
// abort landing in that window would register the abort listener AFTER
|
||||
// the signal fired — never invoked, entry pending forever, zombie frame
|
||||
// on every mux replay. Settle synchronously instead of publishing.
|
||||
if (req.signal?.aborted === true) return Promise.resolve<ApprovalOutcome>('cancelled')
|
||||
// The audit pair `approval/asked` is already appended by the service
|
||||
// before dispatch, but dispatch rides a microtask: parallel tool calls
|
||||
// can append several asked events before any answerer runs. THIS
|
||||
// request's event is therefore the newest asked event that is still
|
||||
// undecided, unclaimed by another pending entry, and — when the ask
|
||||
// names a call — carries the same callId.
|
||||
const events = req.agent.session.events
|
||||
const claimed = new Set<ApprovalRequestId>()
|
||||
for (const entry of pendingApprovals.values()) claimed.add(entry.approvalId)
|
||||
const decided = new Set<ApprovalRequestId>()
|
||||
let approvalId: ApprovalRequestId | undefined
|
||||
for (let i = events.length - 1; i >= 0; i -= 1) {
|
||||
const event = events[i] as SessionEvent
|
||||
if (event.type === 'approval/decided') {
|
||||
decided.add(event.data.id)
|
||||
} else if (event.type === 'approval/asked') {
|
||||
if (decided.has(event.data.id) || claimed.has(event.data.id)) continue
|
||||
// Symmetric pairing: a callId-bearing ask only takes its own call's
|
||||
// record, and a callId-less ask only takes a callId-less record —
|
||||
// so neither shape can steal the other's audit id under parallel
|
||||
// asks. (Today every producer — the tool executor — passes callId;
|
||||
// the callId-less arm guards any future non-tool asker.)
|
||||
if ((req.callId ?? null) !== (event.data.callId ?? null)) continue
|
||||
approvalId = event.data.id
|
||||
break
|
||||
}
|
||||
}
|
||||
// No asked event means the request bypassed the service's audit path —
|
||||
// not this channel's question; delegate to the fail-closed default.
|
||||
if (approvalId === undefined) return next()
|
||||
const id = approvalId
|
||||
return new Promise<ApprovalOutcome>((resolve) => {
|
||||
const settle = (outcome: ApprovalOutcome): void => {
|
||||
/* v8 ignore next 3 -- defensive double-settle guard: respond() routes
|
||||
through the pending table (a settled id is not-pending before it can
|
||||
re-settle) and the first settle removes the abort listener, so no
|
||||
reachable path settles twice; kept against future settle callers. */
|
||||
if (!pendingApprovals.delete(pending.rpcId)) return
|
||||
req.signal?.removeEventListener('abort', onAbort)
|
||||
broadcast({ type: 'approval/resolved', sessionId: pending.sessionId, approvalId: id, outcome })
|
||||
// A cancelled ask was already settled by the service's own signal
|
||||
// race, which discards this late resolution; resolving is a no-op
|
||||
// there and keeps this promise from dangling forever.
|
||||
resolve(outcome)
|
||||
}
|
||||
const onAbort = (): void => { settle('cancelled') }
|
||||
const pending: PendingApproval = {
|
||||
rpcId: RpcId(randomUUID()),
|
||||
sessionId: req.agent.session.id,
|
||||
approvalId: id,
|
||||
toolName: req.toolName,
|
||||
...req.callId === undefined ? {} : { callId: req.callId },
|
||||
...req.reason === undefined ? {} : { reason: req.reason },
|
||||
resolve: settle,
|
||||
}
|
||||
pendingApprovals.set(pending.rpcId, pending)
|
||||
req.signal?.addEventListener('abort', onAbort, { once: true })
|
||||
const envelope = requestedFrame(pending)
|
||||
for (const queue of muxQueues) queue.push(envelope)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Gate the cold path on the store: an id absent from it, or naming a legacy
|
||||
* log without a cwd (pre-release stance: not served, no compatibility), is
|
||||
@@ -671,7 +846,11 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
signal?.throwIfAborted()
|
||||
const items = ctx.sessions.list().map((session) => {
|
||||
const agent = ctx.agents.get(session.id)
|
||||
return summarize(session, agent?.status === 'running')
|
||||
const projections = listProjectionsFor(ctx, session.header, session)
|
||||
return {
|
||||
...summarize(session, agent?.status === 'running'),
|
||||
...projections === undefined ? {} : { projections },
|
||||
}
|
||||
})
|
||||
signal?.throwIfAborted()
|
||||
const attached = new Set(items.map(item => item.sessionId))
|
||||
@@ -684,7 +863,15 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
signal?.throwIfAborted()
|
||||
const batch = cold.slice(offset, offset + COLD_SUMMARY_BATCH_SIZE)
|
||||
const settled = await Promise.allSettled(
|
||||
batch.map(meta => summarizeCold(persistence, meta, signal)),
|
||||
batch.map(async (meta) => {
|
||||
// Cold rows read the persisted projection cache only — never a
|
||||
// log load; a session without a cache row simply has no column.
|
||||
const projections = listProjectionsFor(ctx, meta, undefined)
|
||||
return {
|
||||
...await summarizeCold(persistence, meta, signal),
|
||||
...projections === undefined ? {} : { projections },
|
||||
}
|
||||
}),
|
||||
)
|
||||
const summaries: SessionSummary[] = []
|
||||
let rejected = false
|
||||
@@ -706,6 +893,38 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
return items
|
||||
}
|
||||
|
||||
/** Resolve the goal service; absent = the deployment did not compose @deepseek-ai/dsh-goal. */
|
||||
function goalService(): NonNullable<ReturnType<typeof ctx.get<'goals'>>> | { error: RpcError } {
|
||||
const goals = ctx.get('goals')
|
||||
if (goals === undefined) {
|
||||
return { error: { code: 'internal', message: 'goal service is absent: this deployment does not mount @deepseek-ai/dsh-goal in its composition (cordis.yml or explicit assembly)', details: {} } }
|
||||
}
|
||||
return goals
|
||||
}
|
||||
|
||||
/** Map one goal-domain rejection to the wire error (stable GoalError codes ride in details). */
|
||||
function goalError(request: RpcRequest<unknown>, error: unknown): RpcResponse<never> {
|
||||
const details = error instanceof GoalError ? { goalCode: error.code } : {}
|
||||
return err(request, { code: 'internal', message: String(error), details })
|
||||
}
|
||||
|
||||
/** Resolve a session's agent, apply one goal mutation, and acknowledge with the new CAS ref. */
|
||||
async function mutateGoal(
|
||||
request: RpcRequest<{ sessionId: SessionId }>,
|
||||
mutation: (goals: NonNullable<ReturnType<typeof ctx.get<'goals'>>>, agent: Agent) => CoreGoalRef,
|
||||
): Promise<RpcResponse<{ ref: GoalRef }>> {
|
||||
const goals = goalService()
|
||||
if ('error' in goals) return err(request, goals.error)
|
||||
const found = await agentFor(request.payload.sessionId)
|
||||
if ('error' in found) return err(request, found.error)
|
||||
try {
|
||||
const ref = mutation(goals, found.agent)
|
||||
return ok(request, { ref: { id: ref.id, revision: ref.revision } })
|
||||
} catch (error: unknown) {
|
||||
return goalError(request, error)
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
sessions: {
|
||||
// Attached sessions summarize from memory; persisted-but-unattached (cold)
|
||||
@@ -898,6 +1117,8 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
const { sessionId, beforeSeq, maxMessages } = request.payload
|
||||
const found = await agentFor(sessionId)
|
||||
if ('error' in found) return err(request, found.error)
|
||||
// Everything below the resume above is synchronous: the page slice,
|
||||
// the seq read, and the projection walk see one un-torn session state.
|
||||
const page = paginate(found.agent.session.events, beforeSeq, maxMessages ?? DEFAULT_MAX_MESSAGES)
|
||||
// Views are computed against the registry at pagination time; result
|
||||
// pairing scans within the page only (message-boundary pagination keeps
|
||||
@@ -906,11 +1127,14 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
const view = viewFor(ctx, event, callId => backscanArgs(page.events, callId))
|
||||
return { event, ...view === undefined ? {} : { view } }
|
||||
})
|
||||
// Tail page carries the session-level todo projection over the FULL
|
||||
// log (the page window may not contain the last todo/write; a paged
|
||||
// client cannot reconstruct session-level state from it).
|
||||
const todos = beforeSeq === undefined ? backscanTodos(found.agent.session.events) : undefined
|
||||
return ok(request, { events: entries, hasMore: page.hasMore, ...todos === undefined ? {} : { todos } })
|
||||
// Baseline rider: tail page only — loadOlder (beforeSeq present) is
|
||||
// the one path that never needs a fresh projection baseline.
|
||||
const projections = beforeSeq === undefined ? projectionsFor(ctx, found.agent) : undefined
|
||||
return ok(request, {
|
||||
events: entries,
|
||||
hasMore: page.hasMore,
|
||||
...projections === undefined ? {} : { projections },
|
||||
})
|
||||
},
|
||||
|
||||
async models(request) {
|
||||
@@ -1014,6 +1238,36 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
}
|
||||
},
|
||||
|
||||
async rename(request) {
|
||||
const { sessionId, title } = request.payload
|
||||
const found = await agentFor(sessionId)
|
||||
if ('error' in found) return err(request, found.error)
|
||||
const titles = ctx.get('sessionTitle')
|
||||
if (titles === undefined) {
|
||||
return err(request, { code: 'internal', message: 'renaming is unavailable: this deployment mounts no session-title service', details: {} })
|
||||
}
|
||||
try {
|
||||
const accepted = titles.rename(found.agent.session, title)
|
||||
return ok(request, { title: accepted.title, seq: accepted.eventSeq })
|
||||
} catch (error: unknown) {
|
||||
// Only the input's fault maps to title-invalid (the message is
|
||||
// product-user-visible in the rename dialog); liveness and disposal
|
||||
// races are deployment trouble, not a bad title.
|
||||
if (error instanceof SessionTitleInvalidError) {
|
||||
return err(request, {
|
||||
code: 'title-invalid',
|
||||
message: error.message,
|
||||
details: { sessionId },
|
||||
})
|
||||
}
|
||||
return err(request, {
|
||||
code: 'internal',
|
||||
message: `failed to rename session "${sessionId}": ${String(error)}`,
|
||||
details: {},
|
||||
})
|
||||
}
|
||||
},
|
||||
|
||||
async prompt(request) {
|
||||
const { sessionId, mode, content } = request.payload
|
||||
const found = await agentFor(sessionId)
|
||||
@@ -1022,8 +1276,9 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
// The rpcId rides MessageSource into user/message (merge declaration in api/sessions.ts; provisional correlation).
|
||||
const source: MessageSource = { kind: 'user', rpcId: request.rpcId }
|
||||
try {
|
||||
if (mode === 'steer') agent.steer({ content, source })
|
||||
else agent.followup({ content, source })
|
||||
const message: UserMessage = createUserMessage({ content, source })
|
||||
if (mode === 'steer') agent.steer(message)
|
||||
else agent.followup(message)
|
||||
} catch (error: unknown) {
|
||||
// A synchronous throw from steer/followup means disposed or invalid input; surface as agent-busy with the reason attached.
|
||||
return err(request, { code: 'agent-busy', message: 'prompt rejected', details: { reason: String(error) } })
|
||||
@@ -1181,8 +1436,16 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
},
|
||||
|
||||
async pickDirectory(request, signal) {
|
||||
const capability = ctx.directoryPicker.capability()
|
||||
if (capability.kind !== 'native') {
|
||||
return err(request, {
|
||||
code: 'directory-picker-unavailable',
|
||||
message: `host.pickDirectory needs the native capability; the composed picker serves "${capability.kind}"`,
|
||||
details: { capability: capability.kind },
|
||||
})
|
||||
}
|
||||
try {
|
||||
const path = await (defaults.pickDirectory ?? pickNativeDirectory)(signal)
|
||||
const path = await capability.pick(signal)
|
||||
return ok(request, { path })
|
||||
} catch (error: unknown) {
|
||||
if (signal.aborted) {
|
||||
@@ -1199,6 +1462,67 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
})
|
||||
}
|
||||
},
|
||||
|
||||
async listDirectory(request, signal) {
|
||||
const capability = ctx.directoryPicker.capability()
|
||||
if (capability.kind !== 'browse') {
|
||||
return err(request, {
|
||||
code: 'directory-picker-unavailable',
|
||||
message: `host.listDirectory needs the browse capability; the composed picker serves "${capability.kind}"`,
|
||||
details: { capability: capability.kind },
|
||||
})
|
||||
}
|
||||
try {
|
||||
// The carrier's signal follows the caller: a disconnect or timeout
|
||||
// stops the backend's directory scan instead of outliving it.
|
||||
return ok(request, await capability.list(request.payload.path, signal))
|
||||
} catch (error: unknown) {
|
||||
// An abort is the caller's own timeout/disconnect, not a server
|
||||
// failure — same code pickDirectory and command.execute report.
|
||||
if (signal.aborted) {
|
||||
return err(request, { code: 'cancelled', message: 'directory listing was aborted', details: {} })
|
||||
}
|
||||
return err(request, directoryError(error))
|
||||
}
|
||||
},
|
||||
|
||||
async createDirectory(request) {
|
||||
const capability = ctx.directoryPicker.capability()
|
||||
if (capability.kind !== 'browse') {
|
||||
return err(request, {
|
||||
code: 'directory-picker-unavailable',
|
||||
message: `host.createDirectory needs the browse capability; the composed picker serves "${capability.kind}"`,
|
||||
details: { capability: capability.kind },
|
||||
})
|
||||
}
|
||||
try {
|
||||
return ok(request, { path: await capability.createDirectory(request.payload.path, request.payload.name) })
|
||||
} catch (error: unknown) {
|
||||
return err(request, directoryError(error))
|
||||
}
|
||||
},
|
||||
|
||||
async openPath(request, signal) {
|
||||
try {
|
||||
const open = defaults.openPath
|
||||
?? ((path: string, openSignal: AbortSignal) => openNativePath(path, openSignal))
|
||||
await open(request.payload.path, signal)
|
||||
return ok(request, { opened: true as const })
|
||||
} catch (error: unknown) {
|
||||
if (signal.aborted) {
|
||||
return err(request, {
|
||||
code: 'cancelled',
|
||||
message: 'path open was aborted',
|
||||
details: {},
|
||||
})
|
||||
}
|
||||
return err(request, {
|
||||
code: 'internal',
|
||||
message: `path open failed: ${error instanceof Error ? error.message : String(error)}`,
|
||||
details: {},
|
||||
})
|
||||
}
|
||||
},
|
||||
},
|
||||
|
||||
commands: {
|
||||
@@ -1226,12 +1550,15 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
const found = await agentFor(sessionId)
|
||||
if ('error' in found) return err(request, found.error)
|
||||
try {
|
||||
const result = await commands.execute(found.agent, line, signal)
|
||||
if (result === undefined) return ok(request, { matched: false })
|
||||
return ok(request, {
|
||||
matched: true,
|
||||
result: { kind: result.kind, ...result.text === undefined ? {} : { text: result.text } },
|
||||
})
|
||||
// Pure admission: the executor's durable command/run + command/done
|
||||
// pair (broadcast on the mux stream) carries the outcome; the
|
||||
// response reports whether the line resolved to a handler, plus the
|
||||
// minted pairing id so the issuing client can correlate its request
|
||||
// with the flow node the lifecycle events produce.
|
||||
const execution = await commands.execute(found.agent, line, signal)
|
||||
return ok(request, execution === undefined
|
||||
? { matched: false }
|
||||
: { matched: true, commandId: execution.commandId })
|
||||
} catch (error: unknown) {
|
||||
if (signal.aborted) return err(request, { code: 'cancelled', message: 'command execution was aborted', details: {} })
|
||||
return err(request, { code: 'internal', message: `command failed: ${String(error)}`, details: {} })
|
||||
@@ -1239,6 +1566,54 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
},
|
||||
},
|
||||
|
||||
goals: {
|
||||
// Mutations only — the read side is the 'goal' session projection.
|
||||
// Every verb resolves the session's agent (agentFor: implicit cold
|
||||
// resume, the command.* precedent) and acknowledges with the new CAS
|
||||
// ref; the committed goal/change event carries the whole value to every
|
||||
// client through the projection frames.
|
||||
async create(request) {
|
||||
const { objective, maxGoalRounds } = request.payload
|
||||
return mutateGoal(request, (goals, agent) => goals.create(agent, {
|
||||
objective,
|
||||
...(maxGoalRounds !== undefined ? { maxGoalRounds } : {}),
|
||||
}))
|
||||
},
|
||||
|
||||
async edit(request) {
|
||||
const { ref, objective, maxGoalRounds } = request.payload
|
||||
return mutateGoal(request, (goals, agent) => goals.edit(agent, ref, {
|
||||
...(objective !== undefined ? { objective } : {}),
|
||||
...(maxGoalRounds !== undefined ? { maxGoalRounds } : {}),
|
||||
}))
|
||||
},
|
||||
|
||||
async pause(request) {
|
||||
return mutateGoal(request, (goals, agent) => goals.pause(agent, request.payload.ref))
|
||||
},
|
||||
|
||||
async resume(request) {
|
||||
return mutateGoal(request, (goals, agent) => goals.resume(agent, request.payload.ref))
|
||||
},
|
||||
|
||||
async complete(request) {
|
||||
return mutateGoal(request, (goals, agent) => goals.complete(agent, request.payload.ref))
|
||||
},
|
||||
|
||||
async clear(request) {
|
||||
const goals = goalService()
|
||||
if ('error' in goals) return err(request, goals.error)
|
||||
const found = await agentFor(request.payload.sessionId)
|
||||
if ('error' in found) return err(request, found.error)
|
||||
try {
|
||||
goals.clear(found.agent, request.payload.ref)
|
||||
return ok(request, { cleared: true as const })
|
||||
} catch (error: unknown) {
|
||||
return goalError(request, error)
|
||||
}
|
||||
},
|
||||
},
|
||||
|
||||
skills: {
|
||||
// Skill lookup never touches the Agent registry: the session address
|
||||
// resolves to a canonical cwd from the host-resident session header, so
|
||||
@@ -1269,7 +1644,8 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
return err(request, { code: 'internal', message: 'skill registry is absent: this deployment does not mount @deepseek-ai/dsh-skill in its composition (cordis.yml or explicit assembly)', details: {} })
|
||||
}
|
||||
try {
|
||||
const skills = await skillRegistry.list({ cwd })
|
||||
const skills = (await skillRegistry.list({ cwd }))
|
||||
.filter(skill => skill.invocation.modelInvocable && skill.invocation.userInvocable)
|
||||
return ok(request, {
|
||||
skills: skills.map(skill => ({
|
||||
name: skill.name,
|
||||
@@ -1299,16 +1675,18 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
},
|
||||
})
|
||||
}
|
||||
// Refresh recovery: still-pending approval questions replay with their
|
||||
// stable rpcId so a reconnecting client can still answer them.
|
||||
for (const pending of pendingApprovals.values()) queue.push(requestedFrame(pending))
|
||||
// Queue snapshot baseline (pendingQuestions precedent): frames replayed
|
||||
// in arrival order per session; a reconnecting client rebuilds its
|
||||
// queue view from these alone.
|
||||
for (const [sessionId, entries] of queuedMirror) {
|
||||
for (const entry of entries.values()) {
|
||||
for (const entry of entries) {
|
||||
queue.push(frame({
|
||||
type: 'session/queued',
|
||||
sessionId,
|
||||
content: entry.message.content,
|
||||
source: entry.message.source,
|
||||
message: entry.message,
|
||||
steering: entry.steering,
|
||||
}))
|
||||
}
|
||||
@@ -1334,10 +1712,6 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
const view = viewFor(ctx, event, callId =>
|
||||
openCalls.get(session.id)?.get(callId) ?? backscanArgs(session.events, callId))
|
||||
queue.push(frame({ type: 'session/event', sessionId: session.id, event, ...view === undefined ? {} : { view } }))
|
||||
if (event.type === 'session/title') {
|
||||
// The accepted raw event is already in session.events, so the fold must find it.
|
||||
queue.push(frame(titleFrame(session) as SessionTitleFrame))
|
||||
}
|
||||
}),
|
||||
ctx.on('session/created', (session: Session) => {
|
||||
subscribeSession(queue, session)
|
||||
@@ -1363,8 +1737,8 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
type: 'host/session-added',
|
||||
sessionId: session.id,
|
||||
// Derived at frame time like summarize(); a just-created session
|
||||
// has no events yet, so this is constantly true in practice.
|
||||
blank: session.events.length === 0,
|
||||
// has run no turn yet, so this is constantly true in practice.
|
||||
blank: sessionBlank(session),
|
||||
...session.header.parentSession === undefined ? {} : { parentSessionId: session.header.parentSession },
|
||||
// cwd rides the frame so the client list needs no refresh to group the new session.
|
||||
...session.header.cwd === undefined ? {} : { cwd: session.header.cwd },
|
||||
@@ -1421,6 +1795,20 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
},
|
||||
|
||||
respond(message: ClientResponse): Promise<RpcReceipt> {
|
||||
// Route by the echoed rpcId (the wire correlation): approvals first,
|
||||
// then questions — the two registries share one id space of UUIDs.
|
||||
const approval = pendingApprovals.get(message.rpcId)
|
||||
if (approval !== undefined) {
|
||||
if (!message.result.ok) return Promise.resolve({ accepted: false, reason: 'bad-response' })
|
||||
const parsed = approvalResponsePayloadSchema.safeParse(message.result.value)
|
||||
// The payload's audit correlation must match the entry the rpcId routed
|
||||
// to — a mismatched answer is malformed, not merely late.
|
||||
if (!parsed.success || parsed.data.approvalId !== approval.approvalId || parsed.data.sessionId !== approval.sessionId) {
|
||||
return Promise.resolve({ accepted: false, reason: 'bad-response' })
|
||||
}
|
||||
approval.resolve(parsed.data.outcome)
|
||||
return Promise.resolve({ accepted: true })
|
||||
}
|
||||
const pending = pendingQuestions.get(message.rpcId)
|
||||
if (pending === undefined) return Promise.resolve({ accepted: false, reason: 'not-pending' })
|
||||
if (!message.result.ok) {
|
||||
|
||||
Reference in New Issue
Block a user