fix(apiproxy): serve history events and projections from one log position
Review found two defects in the previous commit's ordering fix. @pku-xht: `historyStateFor` copied the attached session's events, the handler then awaited `presenterScopeFor`, and only then read the projection baseline off the still-live Session. An append during that await served events cut at N beside a baseline folded to N+1 — one response describing two moments. The same restructure had also moved the baseline read outside the `try`, so a failing snapshot escaped the structured `internal` error. Both awaits now happen before the cut: `historySourceFor` resolves which session serves the read, `presenterScopeFor` ensures the recorded composition, and `historyCutOf` then reads events and baseline adjacently with nothing between them. The whole sequence is back inside the try. The invariant judged any scoped assembly with a chain of one as an unjoined agent, which rejects a legitimate assembly in a standing preset key (that key has no parent of its own). It now gates on `context.agent` — a scope-only read is not an agent and is out of range by construction rather than by a premise about who else calls `assemble` — and asks the roster's own `composedPreset()` instead of introspecting chain length. The advisory warning uses the same relation. Also from review: the `2026-08-05-per-agent-tool-presentation` note still described `presentAs` as per-agent, which standing mounts made false and this branch's own rewording contradicts; the duplicated "process-wide unit table" argument collapses to the Agent Note with pointers from the five copies; a dead `.sort()` before `arrayContaining`; and change-history narration in the tool-cordis README.
This commit is contained in:
@@ -686,6 +686,16 @@ function historyPage(
|
||||
* registry). An absent registry means the deployment has no projection seam:
|
||||
* the whole block is absent and clients treat every key as capability-absent.
|
||||
*/
|
||||
/**
|
||||
* Which session a transcript read is served from. An attached session is the
|
||||
* live object and keeps appending, so its events and projection baseline are
|
||||
* read together in one synchronous step; a detached one is already a frozen
|
||||
* inspection.
|
||||
*/
|
||||
type HistorySource =
|
||||
| { readonly kind: 'attached'; readonly session: Session }
|
||||
| { readonly kind: 'detached'; readonly header: SessionHeader; readonly events: SessionEvent[] }
|
||||
|
||||
function projectionsFor(ctx: Context, session: Session): SessionProjectionsBlock | undefined {
|
||||
const registry = ctx.get('sessionProjections')
|
||||
if (registry === undefined) return undefined
|
||||
@@ -1330,39 +1340,54 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
}
|
||||
|
||||
/**
|
||||
* Read one transcript cut without acquiring an Agent owner, plus a DEFERRED
|
||||
* read of the projection baseline.
|
||||
*
|
||||
* The baseline is a thunk rather than a value because the unit table is
|
||||
* process-wide while the units themselves are registered by preset rows: a
|
||||
* key like `todos` exists only once its preset's standing mount is composed.
|
||||
* The caller ensures that mount through {@link presenterScopeFor} — which
|
||||
* needs the header this function returns — so reading the snapshot eagerly
|
||||
* would serve a first cold read a page missing every preset-owned key, and
|
||||
* every later read a complete one.
|
||||
* Resolve which session one transcript read is served from, without
|
||||
* acquiring an Agent owner. This is the read's only asynchronous step
|
||||
* besides ensuring the composition; {@link historyCutOf} takes the cut.
|
||||
* @param sessionId - the transcript being read.
|
||||
* @returns the attached session, or the inspected detached header and events.
|
||||
* @throws {@link ApiRemoteSessionNotFound} when no project-backed session has that identity.
|
||||
*/
|
||||
async function historyStateFor(
|
||||
sessionId: SessionId,
|
||||
includeProjections: boolean,
|
||||
): Promise<{
|
||||
header: SessionHeader
|
||||
events: SessionEvent[]
|
||||
readProjections: () => SessionProjectionsBlock | undefined
|
||||
}> {
|
||||
async function historySourceFor(sessionId: SessionId): Promise<HistorySource> {
|
||||
const attached = ctx.sessions.get(sessionId)
|
||||
if (attached !== undefined) {
|
||||
return {
|
||||
header: attached.header,
|
||||
events: [...attached.events],
|
||||
readProjections: () => includeProjections ? projectionsFor(ctx, attached) : undefined,
|
||||
}
|
||||
}
|
||||
if (attached !== undefined) return { kind: 'attached', session: attached }
|
||||
const inspected = await inspectServable(sessionId)
|
||||
return {
|
||||
header: inspected.meta,
|
||||
events: inspected.events,
|
||||
readProjections: () => includeProjections ? detachedProjectionsFor(ctx, inspected.events) : undefined,
|
||||
return { kind: 'detached', header: inspected.meta, events: inspected.events }
|
||||
}
|
||||
|
||||
/**
|
||||
* The header and events {@link presenterScopeFor} reads to decide which
|
||||
* composition a transcript ran under.
|
||||
* @param source - the live or detached session this read is served from.
|
||||
* @returns that session's creation header and its events.
|
||||
*/
|
||||
function sourceSession(source: HistorySource): PresetBearingSession {
|
||||
if (source.kind === 'detached') return { header: source.header, events: source.events }
|
||||
return { header: source.session.header, events: source.session.events }
|
||||
}
|
||||
|
||||
/**
|
||||
* One transcript cut: the events and the projection baseline that describe
|
||||
* the SAME log position.
|
||||
*
|
||||
* Synchronous, and the two reads sit next to each other, because an attached
|
||||
* session keeps appending: an `await` between them would serve events cut at
|
||||
* N beside a baseline folded to N+1, which is one response describing two
|
||||
* moments. The caller does its awaiting before this call.
|
||||
* @param source - the live or detached session this read is served from.
|
||||
* @param includeProjections - whether the caller asked for the baseline (a tail page does).
|
||||
* @returns the events and, when asked, the baseline for that same position.
|
||||
*/
|
||||
function historyCutOf(
|
||||
source: HistorySource,
|
||||
includeProjections: boolean,
|
||||
): { events: SessionEvent[]; projections?: SessionProjectionsBlock } {
|
||||
if (source.kind === 'detached') {
|
||||
const projections = includeProjections ? detachedProjectionsFor(ctx, source.events) : undefined
|
||||
return { events: source.events, ...projections === undefined ? {} : { projections } }
|
||||
}
|
||||
const events = [...source.session.events]
|
||||
const projections = includeProjections ? projectionsFor(ctx, source.session) : undefined
|
||||
return { events, ...projections === undefined ? {} : { projections } }
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -2022,9 +2047,22 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
|
||||
async history(request) {
|
||||
const { sessionId, beforeSeq, maxMessages } = request.payload
|
||||
let state: Awaited<ReturnType<typeof historyStateFor>>
|
||||
try {
|
||||
state = await historyStateFor(sessionId, beforeSeq === undefined)
|
||||
const source = await historySourceFor(sessionId)
|
||||
// Both awaits happen BEFORE the cut. Ensuring the recorded
|
||||
// composition's standing mount is what registers its projection
|
||||
// units, so a first cold read would otherwise serve a baseline
|
||||
// missing every preset-owned key; and an attached session keeps
|
||||
// appending, so awaiting between the two reads would pair events cut
|
||||
// at N with a baseline folded to N+1.
|
||||
const scope = await presenterScopeFor(sessionId, sourceSession(source))
|
||||
const cut = historyCutOf(source, beforeSeq === undefined)
|
||||
const page = historyPage(ctx, cut.events, beforeSeq, maxMessages, scope)
|
||||
return ok(request, {
|
||||
events: page.events,
|
||||
hasMore: page.hasMore,
|
||||
...cut.projections === undefined ? {} : { projections: cut.projections },
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
if (error instanceof SessionNotFound) {
|
||||
return err(request, { code: 'session-not-found', message: error.message, details: { sessionId } })
|
||||
@@ -2035,17 +2073,6 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
details: {},
|
||||
})
|
||||
}
|
||||
// The scope resolves first: ensuring the recorded composition's
|
||||
// standing mount is what registers its projection units, so the
|
||||
// baseline below has to be read after it, not beside it.
|
||||
const scope = await presenterScopeFor(sessionId, state)
|
||||
const page = historyPage(ctx, state.events, beforeSeq, maxMessages, scope)
|
||||
const projections = state.readProjections()
|
||||
return ok(request, {
|
||||
events: page.events,
|
||||
hasMore: page.hasMore,
|
||||
...projections === undefined ? {} : { projections },
|
||||
})
|
||||
},
|
||||
|
||||
async models(request) {
|
||||
|
||||
Reference in New Issue
Block a user