feat(apiproxy): projection column on session.list — cold titles with zero log loads
SessionSummary grows an optional projections column (whole value per key, same passthrough posture as the history-tail block): attached rows cut the live registry watermark cache; cold rows view the persisted projection cache's stored rows via the new registry viewCheckpoint face (version- matching keys only, zero I/O) — the RFC's motivating scenario, every session's title across a listing without loading one event log. The column is fail-soft and absence-coded: no registry, no cache row, or a throwing read serve the row without the column, never breaking the listing.
This commit is contained in:
@@ -47,6 +47,7 @@
|
|||||||
"@deepseek-ai/dsh-session": "workspace:^",
|
"@deepseek-ai/dsh-session": "workspace:^",
|
||||||
"@deepseek-ai/dsh-session-persistence": "workspace:^",
|
"@deepseek-ai/dsh-session-persistence": "workspace:^",
|
||||||
"@deepseek-ai/dsh-session-projection": "workspace:^",
|
"@deepseek-ai/dsh-session-projection": "workspace:^",
|
||||||
|
"@deepseek-ai/dsh-session-projection-cache": "workspace:^",
|
||||||
"@deepseek-ai/dsh-skill": "workspace:^",
|
"@deepseek-ai/dsh-skill": "workspace:^",
|
||||||
"@deepseek-ai/dsh-tools": "workspace:^",
|
"@deepseek-ai/dsh-tools": "workspace:^",
|
||||||
"@deepseek-ai/dsh-user-approval": "workspace:^",
|
"@deepseek-ai/dsh-user-approval": "workspace:^",
|
||||||
|
|||||||
@@ -29,7 +29,9 @@ import type {
|
|||||||
WorkspaceId, WorkspaceView,
|
WorkspaceId, WorkspaceView,
|
||||||
} from './api/index.ts'
|
} from './api/index.ts'
|
||||||
// Type-only: resolves `ctx.get('sessionProjections')` to the projection registry.
|
// Type-only: resolves `ctx.get('sessionProjections')` to the projection registry.
|
||||||
import type {} from '@deepseek-ai/dsh-session-projection'
|
import type { SessionProjectionMap } 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'
|
||||||
// Type-only edges: resolve `ctx.get('commands')`, the `commands/change` event, and `ctx.get('skills')`.
|
// 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-commands'
|
||||||
import type {} from '@deepseek-ai/dsh-skill'
|
import type {} from '@deepseek-ai/dsh-skill'
|
||||||
@@ -297,6 +299,26 @@ function projectionsFor(ctx: Context, agent: Agent): SessionProjectionsBlock | u
|
|||||||
return registry.snapshot(agent.session)
|
return registry.snapshot(agent.session)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The projection column of one session.list row, fail-soft: attached
|
||||||
|
* sessions cut the registry's live watermark cache; cold sessions view the
|
||||||
|
* persisted projection cache's stored rows (zero log loads either way — the
|
||||||
|
* listing use case the cache exists for). Any failure — and an empty value
|
||||||
|
* set — yields an absent column: a listing without projections is degraded,
|
||||||
|
* never broken.
|
||||||
|
*/
|
||||||
|
function listProjectionsFor(ctx: Context, id: SessionId, session: Session | undefined): Partial<SessionProjectionMap> | undefined {
|
||||||
|
try {
|
||||||
|
const values = session !== undefined
|
||||||
|
? ctx.get('sessionProjections')?.snapshot(session).values
|
||||||
|
: ctx.get('sessionProjectionCache')?.cachedValues(id)
|
||||||
|
return values !== undefined && Object.keys(values).length > 0 ? values : undefined
|
||||||
|
} catch (error) {
|
||||||
|
ctx.logger.warn(`session.list: projection column for "${id}" failed (serving the row without it): ${String(error)}`)
|
||||||
|
return undefined
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Thrown by the cold-resume path when the id names no servable session
|
* Thrown by the cold-resume path when the id names no servable session
|
||||||
* (absent from the store, or a pre-project legacy log without a cwd).
|
* (absent from the store, or a pre-project legacy log without a cwd).
|
||||||
@@ -654,13 +676,25 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
async list(request) {
|
async list(request) {
|
||||||
const items = ctx.sessions.list().map((session) => {
|
const items = ctx.sessions.list().map((session) => {
|
||||||
const agent = ctx.agents.get(session.id)
|
const agent = ctx.agents.get(session.id)
|
||||||
return summarize(session, agent?.status === 'running')
|
const projections = listProjectionsFor(ctx, session.id, session)
|
||||||
|
return {
|
||||||
|
...summarize(session, agent?.status === 'running'),
|
||||||
|
...projections === undefined ? {} : { projections },
|
||||||
|
}
|
||||||
})
|
})
|
||||||
const attached = new Set(items.map(item => item.sessionId))
|
const attached = new Set(items.map(item => item.sessionId))
|
||||||
const persistence = ctx.get('sessionPersistence')
|
const persistence = ctx.get('sessionPersistence')
|
||||||
if (persistence !== undefined) {
|
if (persistence !== undefined) {
|
||||||
const cold = (await persistence.list()).filter(meta => !attached.has(meta.id) && meta.cwd !== undefined)
|
const cold = (await persistence.list()).filter(meta => !attached.has(meta.id) && meta.cwd !== undefined)
|
||||||
items.push(...await Promise.all(cold.map(meta => summarizeCold(persistence, meta))))
|
items.push(...await Promise.all(cold.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.id, undefined)
|
||||||
|
return {
|
||||||
|
...await summarizeCold(persistence, meta),
|
||||||
|
...projections === undefined ? {} : { projections },
|
||||||
|
}
|
||||||
|
})))
|
||||||
}
|
}
|
||||||
items.sort((a, b) => b.updatedAt - a.updatedAt)
|
items.sort((a, b) => b.updatedAt - a.updatedAt)
|
||||||
return ok(request, { items })
|
return ok(request, { items })
|
||||||
|
|||||||
@@ -7,6 +7,7 @@
|
|||||||
|
|
||||||
import { z } from 'zod'
|
import { z } from 'zod'
|
||||||
import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
|
import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
|
||||||
|
import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
|
||||||
import type { RequestPayload, ResponseValue } from './rpc-map.ts'
|
import type { RequestPayload, ResponseValue } from './rpc-map.ts'
|
||||||
import type { Wire } from './rpc.schema.ts'
|
import type { Wire } from './rpc.schema.ts'
|
||||||
import type {
|
import type {
|
||||||
@@ -37,6 +38,15 @@ export const sessionEventSchema = z.object({
|
|||||||
surfaceOp: z.unknown().optional(),
|
surfaceOp: z.unknown().optional(),
|
||||||
}) as unknown as z.ZodType<SessionEvent>
|
}) as unknown as z.ZodType<SessionEvent>
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Projection-values passthrough (same posture as
|
||||||
|
* {@link sessionProjectionsBlockSchema}): each value already passed its
|
||||||
|
* unit's own schema on the host side; deep-validating here would import
|
||||||
|
* every domain's schema into the carrier.
|
||||||
|
*/
|
||||||
|
const projectionValuesSchema =
|
||||||
|
z.record(z.string(), z.unknown()) as unknown as z.ZodType<Partial<SessionProjectionMap>>
|
||||||
|
|
||||||
/** SessionSummary row of session.list. */
|
/** SessionSummary row of session.list. */
|
||||||
export const sessionSummarySchema = z.object({
|
export const sessionSummarySchema = z.object({
|
||||||
sessionId: sessionIdSchema,
|
sessionId: sessionIdSchema,
|
||||||
@@ -45,6 +55,7 @@ export const sessionSummarySchema = z.object({
|
|||||||
blank: z.boolean(),
|
blank: z.boolean(),
|
||||||
parentSessionId: sessionIdSchema.optional(),
|
parentSessionId: sessionIdSchema.optional(),
|
||||||
cwd: z.string().optional(),
|
cwd: z.string().optional(),
|
||||||
|
projections: projectionValuesSchema.optional(),
|
||||||
}) satisfies z.ZodType<Wire<SessionSummary>>
|
}) satisfies z.ZodType<Wire<SessionSummary>>
|
||||||
|
|
||||||
/** session.list request payload (cursor is a reserved seat, unimplemented in v1). */
|
/** session.list request payload (cursor is a reserved seat, unimplemented in v1). */
|
||||||
|
|||||||
@@ -143,6 +143,17 @@ export interface SessionSummary {
|
|||||||
parentSessionId?: SessionId
|
parentSessionId?: SessionId
|
||||||
/** Session working directory (header.cwd passthrough); absent when unrecorded. */
|
/** Session working directory (header.cwd passthrough); absent when unrecorded. */
|
||||||
cwd?: string
|
cwd?: string
|
||||||
|
/**
|
||||||
|
* Whole current value per projection key, with zero log loads: attached
|
||||||
|
* sessions read the registry's live watermark cut; cold sessions read the
|
||||||
|
* persisted projection cache's stored rows — as stale as that session's
|
||||||
|
* last durable checkpoint, never wrong, superseded by the history tail
|
||||||
|
* baseline the moment the session is opened. Absent when no value is
|
||||||
|
* available (no registry, no cache row for a cold session, or a fail-soft
|
||||||
|
* cache read miss); a listing client treats absence as "no title yet",
|
||||||
|
* exactly like a blank session.
|
||||||
|
*/
|
||||||
|
projections?: Partial<SessionProjectionMap>
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Session-domain unary methods (the map keys session.* of RpcMethodMap). */
|
/** Session-domain unary methods (the map keys session.* of RpcMethodMap). */
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ import { z } from 'zod'
|
|||||||
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||||
import { createUserMessage } from '@deepseek-ai/dsh-llm'
|
import { createUserMessage } from '@deepseek-ai/dsh-llm'
|
||||||
import SessionStore from '@deepseek-ai/dsh-session'
|
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||||
import type { Session } from '@deepseek-ai/dsh-session'
|
import type { Session } from '@deepseek-ai/dsh-session'
|
||||||
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
|
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
|
||||||
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
|
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
|
||||||
@@ -125,6 +125,77 @@ describe('session.history projections block', () => {
|
|||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
describe('session.list projections column', () => {
|
||||||
|
it('serves attached rows from the live registry cut', async () => {
|
||||||
|
const { ctx, session } = await harness(true)
|
||||||
|
ctx.sessionProjections.register(lastUserUnit())
|
||||||
|
seedMessages(session, 1)
|
||||||
|
const response = await api(ctx).sessions.list(request({}))
|
||||||
|
if (!response.result.ok) throw new Error('unreachable')
|
||||||
|
const row = response.result.value.items.find(item => item.sessionId === session.id)
|
||||||
|
expect(row?.projections?.['test/last-user']).toEqual({ text: 'm0' })
|
||||||
|
})
|
||||||
|
|
||||||
|
it('omits the column entirely when no registry is mounted', async () => {
|
||||||
|
const { ctx, session } = await harness(false)
|
||||||
|
seedMessages(session, 1)
|
||||||
|
const response = await api(ctx).sessions.list(request({}))
|
||||||
|
if (!response.result.ok) throw new Error('unreachable')
|
||||||
|
const row = response.result.value.items.find(item => item.sessionId === session.id)
|
||||||
|
expect(row).toBeDefined()
|
||||||
|
expect(row !== undefined && 'projections' in row).toBe(false)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('serves cold rows from the persisted projection cache with zero log loads', async () => {
|
||||||
|
const { ctx } = await harness(true)
|
||||||
|
const coldId = SessionId('session-cold-listing')
|
||||||
|
const load = () => { throw new Error('list must not load event logs') }
|
||||||
|
ctx.provide('sessionPersistence', {
|
||||||
|
list: async () => [{ version: 0, id: coldId, createdAt: 5, cwd: '/tmp' }],
|
||||||
|
locate: () => undefined,
|
||||||
|
load,
|
||||||
|
inspect: load,
|
||||||
|
readFrom: load,
|
||||||
|
} as never)
|
||||||
|
ctx.provide('sessionProjectionCache', {
|
||||||
|
cachedValues: (id: unknown) => (id === coldId ? { 'test/last-user': { text: 'cached' } } : {}),
|
||||||
|
} as never)
|
||||||
|
const response = await api(ctx).sessions.list(request({}))
|
||||||
|
if (!response.result.ok) throw new Error('unreachable')
|
||||||
|
const row = response.result.value.items.find(item => item.sessionId === coldId)
|
||||||
|
expect(row?.running).toBe(false)
|
||||||
|
expect(row?.projections?.['test/last-user']).toEqual({ text: 'cached' })
|
||||||
|
})
|
||||||
|
|
||||||
|
it('cold rows without a cache plugin (or without a stored row) just lack the column', async () => {
|
||||||
|
const { ctx } = await harness(true)
|
||||||
|
const coldId = SessionId('session-cold-uncached')
|
||||||
|
ctx.provide('sessionPersistence', {
|
||||||
|
list: async () => [{ version: 0, id: coldId, createdAt: 5, cwd: '/tmp' }],
|
||||||
|
locate: () => undefined,
|
||||||
|
} as never)
|
||||||
|
const response = await api(ctx).sessions.list(request({}))
|
||||||
|
if (!response.result.ok) throw new Error('unreachable')
|
||||||
|
const row = response.result.value.items.find(item => item.sessionId === coldId)
|
||||||
|
expect(row).toBeDefined()
|
||||||
|
expect(row !== undefined && 'projections' in row).toBe(false)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('a throwing column read degrades that row, never the listing', async () => {
|
||||||
|
const { ctx, session } = await harness(true)
|
||||||
|
ctx.sessionProjections.register({
|
||||||
|
...lastUserUnit(),
|
||||||
|
view: () => { throw new Error('unit exploded') },
|
||||||
|
})
|
||||||
|
seedMessages(session, 1)
|
||||||
|
const response = await api(ctx).sessions.list(request({}))
|
||||||
|
if (!response.result.ok) throw new Error('unreachable')
|
||||||
|
const row = response.result.value.items.find(item => item.sessionId === session.id)
|
||||||
|
expect(row).toBeDefined()
|
||||||
|
expect(row !== undefined && 'projections' in row).toBe(false)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
||||||
describe('session/projection push frame', () => {
|
describe('session/projection push frame', () => {
|
||||||
/** Drain frames until `count` session/projection frames arrived. */
|
/** Drain frames until `count` session/projection frames arrived. */
|
||||||
async function collect(iterable: AsyncIterable<RpcRequest<MuxFrame>>, count: number, abort: AbortController): Promise<MuxFrame[]> {
|
async function collect(iterable: AsyncIterable<RpcRequest<MuxFrame>>, count: number, abort: AbortController): Promise<MuxFrame[]> {
|
||||||
|
|||||||
@@ -35,6 +35,9 @@
|
|||||||
{
|
{
|
||||||
"path": "../../session-projection/session-projection"
|
"path": "../../session-projection/session-projection"
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
"path": "../../session-projection/session-projection-cache"
|
||||||
|
},
|
||||||
{
|
{
|
||||||
"path": "../../skill/skill"
|
"path": "../../skill/skill"
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
|||||||
// Empty type import: applies the package's cordis Context merge
|
// Empty type import: applies the package's cordis Context merge
|
||||||
// (`ctx.sessionPersistence`), which this service reads on the cold path.
|
// (`ctx.sessionPersistence`), which this service reads on the cold path.
|
||||||
import type {} from '@deepseek-ai/dsh-session-persistence'
|
import type {} from '@deepseek-ai/dsh-session-persistence'
|
||||||
import type { ProjectionCheckpoint, ProjectionSnapshot } from '@deepseek-ai/dsh-session-projection'
|
import type { ProjectionCheckpoint, ProjectionSnapshot, SessionProjectionMap } from '@deepseek-ai/dsh-session-projection'
|
||||||
import type { KvTable } from '@deepseek-ai/dsh-storage-domain'
|
import type { KvTable } from '@deepseek-ai/dsh-storage-domain'
|
||||||
import { projectionCacheDomainSpec } from './spec.ts'
|
import { projectionCacheDomainSpec } from './spec.ts'
|
||||||
import type { CheckpointRecord } from './spec.ts'
|
import type { CheckpointRecord } from './spec.ts'
|
||||||
@@ -98,6 +98,20 @@ export class SessionProjectionCache extends Service {
|
|||||||
return this.requireTable().get(id)?.rows ?? {}
|
return this.requireTable().get(id)?.rows ?? {}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The zero-I/O listing read: whole values viewed straight from the stored
|
||||||
|
* rows (version-matching keys only), as stale as the last durable
|
||||||
|
* checkpoint but never wrong. Synchronous — a listing over every stored
|
||||||
|
* session touches no log. Fresher paths (the history tail baseline,
|
||||||
|
* {@link coldSnapshot}) supersede these values whenever a session is
|
||||||
|
* actually opened.
|
||||||
|
* @param id - the session whose cached values are viewed.
|
||||||
|
* @returns whole values per key with a usable row; empty when none stored.
|
||||||
|
*/
|
||||||
|
cachedValues(id: SessionId): Partial<SessionProjectionMap> {
|
||||||
|
return this.ctx.sessionProjections.viewCheckpoint(this.checkpointOf(id))
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Durably checkpoint one live session NOW (both mandatory points call
|
* Durably checkpoint one live session NOW (both mandatory points call
|
||||||
* this; tests and carriers may too). The registry cut is snapshotted at
|
* this; tests and carriers may too). The registry cut is snapshotted at
|
||||||
|
|||||||
@@ -280,6 +280,27 @@ export class SessionProjectionRegistry extends Service {
|
|||||||
return floor === undefined ? undefined : Math.max(floor - 1, 0)
|
return floor === undefined ? undefined : Math.max(floor - 1, 0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* View a checkpoint's rows without any log read: for every registered
|
||||||
|
* unit whose row's `stateVersion` matches, serve the schema-validated
|
||||||
|
* `view` of the stored state; mismatched or absent rows leave their key
|
||||||
|
* absent (a cold or listing consumer treats it as not-yet-available and a
|
||||||
|
* fuller read path refolds it). The zero-I/O rung of the read ladder —
|
||||||
|
* values are as stale as their rows, never wrong.
|
||||||
|
* @param checkpoint - persisted rows for one session (possibly stale or empty).
|
||||||
|
* @returns whole values per key with a usable row; empty when none.
|
||||||
|
*/
|
||||||
|
viewCheckpoint(checkpoint: ProjectionCheckpoint): Partial<SessionProjectionMap> {
|
||||||
|
const values: Record<string, unknown> = {}
|
||||||
|
for (const registration of this.registrations.values()) {
|
||||||
|
const def = registration.def
|
||||||
|
const row = checkpoint[def.key]
|
||||||
|
if (row === undefined || row.stateVersion !== def.stateVersion) continue
|
||||||
|
values[def.key] = def.schema.parse(def.view(row.state))
|
||||||
|
}
|
||||||
|
return values as Partial<SessionProjectionMap>
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Cold read: fold every registered unit over a stored log suffix, seeding
|
* Cold read: fold every registered unit over a stored log suffix, seeding
|
||||||
* each from its checkpoint row when usable — the one read recipe (cached
|
* each from its checkpoint row when usable — the one read recipe (cached
|
||||||
|
|||||||
@@ -267,6 +267,19 @@ describe('SessionProjectionRegistry drive', () => {
|
|||||||
expect(current.values['test/count']).toBe(5)
|
expect(current.values['test/count']).toBe(5)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('viewCheckpoint serves version-matching rows without any log and skips mismatched keys', async () => {
|
||||||
|
const { ctx } = await harness()
|
||||||
|
ctx.sessionProjections.register(marksUnit())
|
||||||
|
ctx.sessionProjections.register(countUnit())
|
||||||
|
const values = ctx.sessionProjections.viewCheckpoint({
|
||||||
|
'test/marks': { stateVersion: 1, observedSeq: 4, state: { marks: ['stored'] } },
|
||||||
|
'test/count': { stateVersion: 99, observedSeq: 4, state: 5 }, // mismatched: absent
|
||||||
|
})
|
||||||
|
expect(values['test/marks']).toEqual({ marks: ['stored'] })
|
||||||
|
expect('test/count' in values).toBe(false)
|
||||||
|
expect(ctx.sessionProjections.viewCheckpoint({})).toEqual({})
|
||||||
|
})
|
||||||
|
|
||||||
it('restore rejects a row claiming events past the supplied log end (shrunk log ⇒ re-read)', async () => {
|
it('restore rejects a row claiming events past the supplied log end (shrunk log ⇒ re-read)', async () => {
|
||||||
const { ctx } = await harness()
|
const { ctx } = await harness()
|
||||||
ctx.sessionProjections.register(countUnit())
|
ctx.sessionProjections.register(countUnit())
|
||||||
|
|||||||
3
pnpm-lock.yaml
generated
3
pnpm-lock.yaml
generated
@@ -2685,6 +2685,9 @@ importers:
|
|||||||
'@deepseek-ai/dsh-session-projection':
|
'@deepseek-ai/dsh-session-projection':
|
||||||
specifier: workspace:^
|
specifier: workspace:^
|
||||||
version: link:../../session-projection/session-projection
|
version: link:../../session-projection/session-projection
|
||||||
|
'@deepseek-ai/dsh-session-projection-cache':
|
||||||
|
specifier: workspace:^
|
||||||
|
version: link:../../session-projection/session-projection-cache
|
||||||
'@deepseek-ai/dsh-skill':
|
'@deepseek-ai/dsh-skill':
|
||||||
specifier: workspace:^
|
specifier: workspace:^
|
||||||
version: link:../../skill/skill
|
version: link:../../skill/skill
|
||||||
|
|||||||
Reference in New Issue
Block a user