You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/09/05 22:08:06 UTC
(maka) 04/06: refactor(runtime): read a running turn from its own ledger
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch refactor/4791-single-transcript-authority
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 863797a0a516b13154f20355b87fc4081f48015c
Author: AstroHan <le...@outlook.com>
AuthorDate: Sun Sep 6 02:40:01 2026 +0800
refactor(runtime): read a running turn from its own ledger
A running turn's rows came from the Session transcript store while every
finished turn's came from the RuntimeEvent ledger. That is the double
write: the same execution facts written twice so a reader could find them
in whichever place it looked.
Now one place answers. An open invocation is read the way the Host's
active overlay already read it -- arriving text presented as settled, a
step that has only thought given the empty assistant row that thinking
hangs on -- and that reading moves next to the projection so both readers
share it instead of keeping a copy each. "Still running" is the absence of
the terminal event, so it is stated on the turn record where it belongs
rather than as a transcript row.
Generated-by: Claude Code
---
.../src/server/execution-composition.ts | 1 -
.../src/server/session-transcript-reader.ts | 66 +--------
.../runtime/src/__tests__/session-manager.test.ts | 108 +++++++-------
packages/runtime/src/runtime-event-read-model.ts | 62 ++++++++
packages/runtime/src/runtime-read-model.ts | 165 ++++-----------------
packages/runtime/src/session-manager.ts | 15 +-
6 files changed, 159 insertions(+), 258 deletions(-)
diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts
index 7c0f8f90e0..7471edb368 100644
--- a/packages/runtime-host/src/server/execution-composition.ts
+++ b/packages/runtime-host/src/server/execution-composition.ts
@@ -966,7 +966,6 @@ export async function createExecutionRuntimeHostComposition(
}),
readModel: new RuntimeReadModel({
runtimeEventStore: stores.runtimeEventStore,
- projectionCache: stores.sessionStore,
canonicalPermissionOutcomes,
}),
artifacts: openedArtifactStore,
diff --git a/packages/runtime-host/src/server/session-transcript-reader.ts b/packages/runtime-host/src/server/session-transcript-reader.ts
index cc51d9bd4a..6eb0259ea5 100644
--- a/packages/runtime-host/src/server/session-transcript-reader.ts
+++ b/packages/runtime-host/src/server/session-transcript-reader.ts
@@ -20,6 +20,7 @@
import type { RuntimeEvent } from '@maka/core/runtime-event';
import type { StoredMessage } from '@maka/core/session';
import {
+ activePresentationRuntimeEvents,
affectsRuntimeEventStoredMessageProjection,
isHardRuntimeEventReadModelDiagnostic,
projectRuntimeEventsToStoredMessages,
@@ -66,10 +67,13 @@ export function createSessionTranscriptReader(input: {
events,
input.canonicalPermissionOutcomes,
);
- const projected = projectRuntimeEventsToStoredMessages(activePresentationEvents(events), {
- invocations: invocations.filter((invocation) => invocation.runId === rootTurn.runId),
- canonicalPermissionOutcomes,
- });
+ const projected = projectRuntimeEventsToStoredMessages(
+ activePresentationRuntimeEvents(events),
+ {
+ invocations: invocations.filter((invocation) => invocation.runId === rootTurn.runId),
+ canonicalPermissionOutcomes,
+ },
+ );
if (projected.diagnostics.some(isHardRuntimeEventReadModelDiagnostic)) {
throw new Error('Active RuntimeEvent transcript projection is incomplete');
}
@@ -144,37 +148,6 @@ async function readCanonicalPermissionOutcomes(
return outcomes;
}
-function activePresentationEvents(events: readonly RuntimeEvent[]): RuntimeEvent[] {
- const textMessages = new Set<string>();
- const lastThinkingByMessage = new Map<string, RuntimeEvent>();
-
- for (const event of events) {
- const content = event.content;
- if (event.role !== 'model' || (content?.kind !== 'text' && content?.kind !== 'thinking')) {
- continue;
- }
- const messageKey = activeMessageKey(event);
- if (content.kind === 'text') textMessages.add(messageKey);
- else lastThinkingByMessage.set(messageKey, event);
- }
-
- const syntheticAfter = new Map<RuntimeEvent, RuntimeEvent[]>();
- for (const [messageKey, thinking] of lastThinkingByMessage) {
- if (textMessages.has(messageKey)) continue;
- const existing = syntheticAfter.get(thinking) ?? [];
- existing.push(emptyAssistantText(thinking));
- syntheticAfter.set(thinking, existing);
- }
-
- const presented: RuntimeEvent[] = [];
- for (const event of events) {
- presented.push(presentationEvent(event));
- const synthetic = syntheticAfter.get(event);
- if (synthetic) presented.push(...synthetic);
- }
- return presented;
-}
-
async function readActiveProjectionEvents(
stores: ExecutionStoresWriter<'interactive'>,
sessionId: string,
@@ -215,29 +188,6 @@ async function readActiveProjectionEvents(
return events;
}
-function activeMessageKey(event: RuntimeEvent): string {
- const messageId = event.refs?.providerEventId ?? event.refs?.storedMessageId ?? event.id;
- return `${event.runId}\0${messageId}`;
-}
-
-function presentationEvent(event: RuntimeEvent): RuntimeEvent {
- const content = event.content;
- return event.partial &&
- event.role === 'model' &&
- (content?.kind === 'text' || content?.kind === 'thinking')
- ? { ...event, partial: false }
- : event;
-}
-
-function emptyAssistantText(thinking: RuntimeEvent): RuntimeEvent {
- return {
- ...thinking,
- id: `${thinking.id}:active-transcript-empty-text`,
- partial: false,
- content: { kind: 'text', text: '' },
- };
-}
-
function isTerminalTurn(turn: TurnSnapshot): boolean {
return turn.status === 'completed' || turn.status === 'failed' || turn.status === 'cancelled';
}
diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts
index e950c2bf04..577e640b15 100644
--- a/packages/runtime/src/__tests__/session-manager.test.ts
+++ b/packages/runtime/src/__tests__/session-manager.test.ts
@@ -7589,17 +7589,6 @@ describe('SessionManager permission mode updates', () => {
hint: 'write approval',
},
);
-
- const cachedView = await new RuntimeReadModel({
- runtimeEventStore: runStore,
- projectionCache: {
- readMessages: async () =>
- messages.filter((message) => message.type !== 'permission_decision'),
- },
- canonicalPermissionOutcomes,
- }).getSessionView(header.sessionId);
-
- assert.deepStrictEqual(cachedView.diagnostics, []);
});
test('SessionManager joins a canonical hosted permission without a ledger request', async () => {
@@ -8221,7 +8210,7 @@ describe('SessionManager permission mode updates', () => {
);
});
- test('getMessages includes in-flight projection cache rows for an active RuntimeEvent run', async () => {
+ test('getMessages reads an active run from its own ledger', async () => {
const store = new MemorySessionStore();
const runStore = new MemoryAgentRunStore();
const manager = makeManagerForReadCutover(store, runStore);
@@ -8236,7 +8225,51 @@ describe('SessionManager permission mode updates', () => {
assistantText: 'completed answer',
legacyIdPrefix: 'legacy',
});
- const activeMessages: StoredMessage[] = [
+ const activeHeader = makeRunHeader({
+ sessionId: session.id,
+ runId: 'run-2',
+ turnId: 'turn-2',
+ status: 'running',
+ createdAt: 200,
+ updatedAt: 203,
+ });
+ await seedInvocationFromHeader(runStore, activeHeader);
+ await runStore.appendRuntimeEvent(
+ session.id,
+ 'run-2',
+ runtimeEvent({
+ id: 'active-user-event',
+ sessionId: session.id,
+ runId: 'run-2',
+ turnId: 'turn-2',
+ ts: 201,
+ role: 'user',
+ author: 'user',
+ content: { kind: 'text', text: 'active question' },
+ refs: { storedMessageId: 'active-user' },
+ }),
+ );
+ // Still arriving: the row a reader sees now, with more of it to come.
+ await runStore.appendRuntimeEvent(
+ session.id,
+ 'run-2',
+ runtimeEvent({
+ id: 'active-assistant-event',
+ sessionId: session.id,
+ runId: 'run-2',
+ turnId: 'turn-2',
+ ts: 202,
+ partial: true,
+ role: 'model',
+ author: 'agent',
+ content: { kind: 'text', text: 'partial active answer' },
+ refs: { storedMessageId: 'active-assistant' },
+ }),
+ );
+
+ const messages = await manager.getMessages(session.id);
+ assert.deepStrictEqual(messages, [
+ ...completed.projectedMessages,
{ type: 'user', id: 'active-user', turnId: 'turn-2', ts: 201, text: 'active question' },
{
type: 'assistant',
@@ -8246,30 +8279,7 @@ describe('SessionManager permission mode updates', () => {
text: 'partial active answer',
modelId: 'fake-model',
},
- {
- type: 'turn_state',
- id: 'active-state',
- turnId: 'turn-2',
- ts: 203,
- status: 'running',
- partialOutputRetained: true,
- },
- ];
- await store.appendMessages(session.id, activeMessages);
- await seedInvocationFromHeader(
- runStore,
- makeRunHeader({
- sessionId: session.id,
- runId: 'run-2',
- turnId: 'turn-2',
- status: 'running',
- createdAt: 200,
- updatedAt: 203,
- }),
- );
-
- const messages = await manager.getMessages(session.id);
- assert.deepStrictEqual(messages, [...completed.projectedMessages, ...activeMessages]);
+ ]);
assert.deepStrictEqual(await manager.listTurns(session.id), [
{
turnId: 'turn-1',
@@ -8284,19 +8294,6 @@ describe('SessionManager permission mode updates', () => {
partialOutputRetained: true,
},
]);
-
- const view = await new RuntimeReadModel({
- runtimeEventStore: runStore,
- projectionCache: store,
- }).getSessionView(session.id);
- assert.strictEqual(
- view.diagnostics.some(
- (diagnostic) =>
- diagnostic.code === 'incomplete_event' &&
- diagnostic.message.includes('in-flight projection cache'),
- ),
- true,
- );
});
test('getMessages overlays a canonical permission acceptance from a running ledger', async () => {
@@ -8480,7 +8477,6 @@ describe('SessionManager permission mode updates', () => {
const view = await new RuntimeReadModel({
runtimeEventStore: runStore,
- projectionCache: store,
}).getSessionView(session.id);
const readRequestId = (value: unknown): string[] =>
@@ -8677,8 +8673,8 @@ describe('SessionManager permission mode updates', () => {
userText: 'runtime regenerate text',
assistantText: 'runtime answer',
legacyIdPrefix: 'legacy',
+ legacyUserText: 'stale transcript text',
});
- store.failNextReadMessagesFor.set(session.id, 1);
await drain(manager.regenerateTurn(session.id, { sourceTurnId: 'source', turnId: 'regen-1' }));
@@ -8812,8 +8808,8 @@ describe('SessionManager permission mode updates', () => {
}),
],
);
- store.failNextReadMessagesFor.set(session.id, 1);
-
+ // The transcript store holds no source rows at all, so what regenerate
+ // finds can only have come from the ledger.
await drain(
manager.regenerateTurn(session.id, { sourceTurnId: 'source', turnId: 'regen-aborted' }),
);
@@ -14228,6 +14224,8 @@ async function seedRuntimeReadTurn(input: {
userText: string;
assistantText: string;
legacyIdPrefix: string;
+ /** Says something else in the transcript store, so a reader proves its source. */
+ legacyUserText?: string;
}): Promise<{ legacyMessages: StoredMessage[]; projectedMessages: StoredMessage[] }> {
const header = makeRunHeader({
sessionId: input.sessionId,
@@ -14279,7 +14277,7 @@ async function seedRuntimeReadTurn(input: {
id: `${input.legacyIdPrefix}-user`,
turnId: input.turnId,
ts: 101,
- text: input.userText,
+ text: input.legacyUserText ?? input.userText,
},
{
type: 'assistant',
diff --git a/packages/runtime/src/runtime-event-read-model.ts b/packages/runtime/src/runtime-event-read-model.ts
index 4dacb25268..9aa5f21deb 100644
--- a/packages/runtime/src/runtime-event-read-model.ts
+++ b/packages/runtime/src/runtime-event-read-model.ts
@@ -440,6 +440,68 @@ export function projectRuntimeEventsToStoredMessages(
return { messages, diagnostics: state.diagnostics };
}
+/**
+ * A running invocation's events as the transcript should show them right now.
+ *
+ * Two things separate a live run from a finished one. Its last text or thinking
+ * event is still arriving, so it is presented as settled rather than withheld;
+ * and a step that has only thought so far has no assistant row to hang that
+ * thinking on, so an empty one is opened for it. Neither changes the ledger:
+ * both are how the same events read before the run ends.
+ */
+export function activePresentationRuntimeEvents(events: readonly RuntimeEvent[]): RuntimeEvent[] {
+ const textMessages = new Set<string>();
+ const lastThinkingByMessage = new Map<string, RuntimeEvent>();
+
+ for (const event of events) {
+ const content = event.content;
+ if (event.role !== 'model' || (content?.kind !== 'text' && content?.kind !== 'thinking')) {
+ continue;
+ }
+ const messageKey = activeMessageKey(event);
+ if (content.kind === 'text') textMessages.add(messageKey);
+ else lastThinkingByMessage.set(messageKey, event);
+ }
+
+ const syntheticAfter = new Map<RuntimeEvent, RuntimeEvent[]>();
+ for (const [messageKey, thinking] of lastThinkingByMessage) {
+ if (textMessages.has(messageKey)) continue;
+ const existing = syntheticAfter.get(thinking) ?? [];
+ existing.push(emptyAssistantText(thinking));
+ syntheticAfter.set(thinking, existing);
+ }
+
+ const presented: RuntimeEvent[] = [];
+ for (const event of events) {
+ presented.push(settledPresentationEvent(event));
+ presented.push(...(syntheticAfter.get(event) ?? []));
+ }
+ return presented;
+}
+
+function activeMessageKey(event: RuntimeEvent): string {
+ const messageId = event.refs?.providerEventId ?? event.refs?.storedMessageId ?? event.id;
+ return `${event.runId}\0${messageId}`;
+}
+
+function settledPresentationEvent(event: RuntimeEvent): RuntimeEvent {
+ const content = event.content;
+ return event.partial &&
+ event.role === 'model' &&
+ (content?.kind === 'text' || content?.kind === 'thinking')
+ ? { ...event, partial: false }
+ : event;
+}
+
+function emptyAssistantText(thinking: RuntimeEvent): RuntimeEvent {
+ return {
+ ...thinking,
+ id: `${thinking.id}:active-transcript-empty-text`,
+ partial: false,
+ content: { kind: 'text', text: '' },
+ };
+}
+
export function projectRuntimeEventsToStoredMessagesWithArchiveStatuses(
events: readonly RuntimeEvent[],
options: ProjectRuntimeEventsToStoredMessagesOptions & {
diff --git a/packages/runtime/src/runtime-read-model.ts b/packages/runtime/src/runtime-read-model.ts
index e103ae0dd5..eb9b8ae98d 100644
--- a/packages/runtime/src/runtime-read-model.ts
+++ b/packages/runtime/src/runtime-read-model.ts
@@ -28,6 +28,7 @@ import type {
CanonicalPermissionOutcomeRecord,
} from './interaction-authority.js';
import {
+ activePresentationRuntimeEvents,
classifyRuntimeEventTerminalFact,
compareRuntimeReadModelMessages,
isHardRuntimeEventReadModelDiagnostic,
@@ -42,13 +43,8 @@ import {
const CANONICAL_PERMISSION_READ_CONCURRENCY = 8;
-export interface RuntimeReadModelProjectionCache {
- readMessages(sessionId: string): Promise<StoredMessage[]>;
-}
-
export interface RuntimeReadModelDeps {
runtimeEventStore: RuntimeEventStore;
- projectionCache?: RuntimeReadModelProjectionCache;
canonicalPermissionOutcomes?: CanonicalPermissionOutcomeReader;
}
@@ -132,29 +128,12 @@ export class RuntimeReadModel {
}
// No terminal event yet: the invocation is still open, or the process died
- // holding it. Either way the ledger is the whole truth about it, so the
- // in-flight projection cache supplies the rows a live turn has not
- // committed instead of a status field claiming otherwise.
+ // holding it. Either way its own events are the whole truth about it, read
+ // as a running turn reads — the arriving text presented as settled. No
+ // durable ordinals exist for them yet, so they keep ledger order.
if (!invocation.terminalEvent) {
- diagnostics.push(
- readModelDiagnostic(
- 'incomplete_event',
- 'active run is using the in-flight projection cache',
- { runId: invocation.runId, turnId: invocation.turnId },
- ),
- );
inFlightTurnIds.add(invocation.turnId);
- if (!this.deps.projectionCache) {
- throw new RuntimeReadModelError('RuntimeEvent ledger is incomplete for an active run', [
- readModelDiagnostic(
- 'incomplete_event',
- 'active run has no stable RuntimeEvent read projection',
- { runId: invocation.runId, turnId: invocation.turnId },
- ),
- ]);
- }
- const overlayEvents = runEvents.flatMap(activeInteractionOverlayEvent);
- appendOrderedEvents(ordered, overlayEvents, runIndex);
+ appendOrderedEvents(ordered, activePresentationRuntimeEvents(runEvents), runIndex);
continue;
}
@@ -222,46 +201,12 @@ export class RuntimeReadModel {
throw new RuntimeReadModelError('RuntimeEvent read projection is incomplete', diagnostics);
}
- const sessionId = input.invocations[0]?.sessionId;
- let cachedMessages: StoredMessage[] | undefined;
- if (sessionId && this.deps.projectionCache) {
- try {
- cachedMessages = await this.deps.projectionCache.readMessages(sessionId);
- } catch (error) {
- const diagnostic = readModelDiagnostic(
- 'unsupported_event',
- 'SessionProjectionCache.readMessages failed',
- {
- error: errorMessage(error),
- },
- );
- diagnostics.push(diagnostic);
- if (input.inFlightTurnIds && input.inFlightTurnIds.size > 0) {
- throw new RuntimeReadModelError(
- 'RuntimeEvent active projection cache read failed',
- diagnostics,
- );
- }
- }
- }
-
- const messages =
- input.inFlightTurnIds && input.inFlightTurnIds.size > 0
- ? mergeInFlightProjectionCache(
- projected.messages,
- cachedMessages ?? [],
- input.inFlightTurnIds,
- )
- : projected.messages;
-
- diagnostics.push(
- ...this.compareProjectionCache(messages, cachedMessages, canonicalPermissionRead.outcomes),
- );
+ const messages = projected.messages;
return {
source: 'runtime_events',
messages,
- turns: deriveTurnRecords(messages),
+ turns: runningTurnRecords(deriveTurnRecords(messages), input.inFlightTurnIds),
events: input.events,
invocations: input.invocations,
diagnostics,
@@ -315,82 +260,36 @@ export class RuntimeReadModel {
);
return { outcomes, diagnostics };
}
-
- private compareProjectionCache(
- messages: readonly StoredMessage[],
- cached: readonly StoredMessage[] | undefined,
- canonicalPermissionOutcomes: ReadonlyMap<string, CanonicalPermissionOutcomeRecord>,
- ): RuntimeEventReadModelDiagnostic[] {
- if (!cached) return [];
- const canonicalRequestIds = new Set(canonicalPermissionOutcomes.keys());
- const excludesCanonicalPermission = (message: StoredMessage): boolean =>
- message.type === 'permission_decision' && canonicalRequestIds.has(message.id);
- return compareRuntimeReadModelMessages(
- messages.filter((message) => !excludesCanonicalPermission(message)),
- cached.filter((message) => !excludesCanonicalPermission(message)),
- ).diagnostics;
- }
}
/**
- * The interaction facts an active run must keep even while its messages come
- * from the in-flight projection cache. Permission prompts were always carried
- * here; sandbox boundary requests and decisions belong for the same reason
- * (#1612): they are the only durable record that a prompt was raised and how
- * it settled, so dropping them makes a pending request invisible to anything
- * reading the view instead of the live backend.
+ * A turn whose invocation has not ended is running.
+ *
+ * The transcript has no row that says so, and it should not: "still running" is
+ * the absence of the terminal event, read off the invocation itself. Rows are
+ * what the turn produced, and a turn that has produced an answer but not ended
+ * would otherwise read as finished.
*/
-function activeInteractionOverlayEvent(event: RuntimeEvent): RuntimeEvent[] {
- const permissionRequest = event.actions?.permissionRequest;
- const permissionAnswerAccepted = event.actions?.permissionAnswerAccepted;
- const permissionClosureAccepted = event.actions?.permissionClosureAccepted;
- const sandboxBoundaryRequest = event.actions?.stateDelta?.sandboxBoundaryRequest;
- const sandboxBoundaryDecision = event.actions?.stateDelta?.sandboxBoundaryDecision;
- if (
- !permissionRequest &&
- !permissionAnswerAccepted &&
- !permissionClosureAccepted &&
- sandboxBoundaryRequest === undefined &&
- sandboxBoundaryDecision === undefined
- ) {
- return [];
- }
- const overlay = { ...event };
- delete overlay.content;
- delete overlay.status;
- const stateDelta = {
- ...(sandboxBoundaryRequest !== undefined ? { sandboxBoundaryRequest } : {}),
- ...(sandboxBoundaryDecision !== undefined ? { sandboxBoundaryDecision } : {}),
- };
- overlay.actions = {
- ...(permissionRequest ? { permissionRequest } : {}),
- ...(permissionAnswerAccepted ? { permissionAnswerAccepted } : {}),
- ...(permissionClosureAccepted ? { permissionClosureAccepted } : {}),
- ...(Object.keys(stateDelta).length > 0 ? { stateDelta } : {}),
- };
- return [overlay];
-}
-
-function mergeInFlightProjectionCache(
- runtimeMessages: readonly StoredMessage[],
- cachedMessages: readonly StoredMessage[],
- inFlightTurnIds: ReadonlySet<string>,
-): StoredMessage[] {
- const merged = runtimeMessages.map((message, index) => ({ message, index }));
- const seenIds = new Set(runtimeMessages.map((message) => message.id));
- for (const cached of cachedMessages) {
- const turnId = messageTurnId(cached);
- if (!turnId || !inFlightTurnIds.has(turnId) || seenIds.has(cached.id)) continue;
- seenIds.add(cached.id);
- merged.push({ message: cached, index: merged.length });
+function runningTurnRecords(
+ turns: readonly TurnRecord[],
+ inFlightTurnIds: ReadonlySet<string> | undefined,
+): TurnRecord[] {
+ if (!inFlightTurnIds || inFlightTurnIds.size === 0) return [...turns];
+ const running = new Set(inFlightTurnIds);
+ const marked = turns.map((turn) => {
+ if (!running.delete(turn.turnId)) return turn;
+ return { ...turn, status: 'running' as const, statusSource: 'recorded' as const };
+ });
+ // An invocation that has opened but produced nothing yet still has a turn.
+ for (const turnId of running) {
+ marked.push({
+ turnId,
+ status: 'running',
+ statusSource: 'recorded',
+ partialOutputRetained: false,
+ });
}
- return merged
- .sort((a, b) => a.message.ts - b.message.ts || a.index - b.index)
- .map((entry) => entry.message);
-}
-
-function messageTurnId(message: StoredMessage): string | undefined {
- return 'turnId' in message && typeof message.turnId === 'string' ? message.turnId : undefined;
+ return marked;
}
function readModelDiagnostic(
diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts
index 38ac4474e2..408e8b2a11 100644
--- a/packages/runtime/src/session-manager.ts
+++ b/packages/runtime/src/session-manager.ts
@@ -150,7 +150,6 @@ import {
import {
RuntimeReadModel,
RuntimeReadModelError,
- type RuntimeReadModelProjectionCache,
type RuntimeReadModelSessionView,
} from './runtime-read-model.js';
import { inspectAgentRunReadModel, type AgentRunInspectModel } from './agent-run-inspect.js';
@@ -3989,7 +3988,7 @@ export class SessionManager {
throw new Error('Conversation copy requires a side-effect-free message snapshot');
}
const readMessages = readMessagesSnapshot.bind(this.deps.store);
- const view = await this.getSessionView(sessionId, { readMessages });
+ const view = await this.getSessionView(sessionId);
if (view.invocations.length > 0 || view.messages.length > 0) return view;
const messages = await readMessages(sessionId);
if (messages.length === 0) return view;
@@ -4341,22 +4340,16 @@ export class SessionManager {
return turn;
}
- private async getSessionView(
- sessionId: string,
- projectionCache: RuntimeReadModelProjectionCache = this.deps.store,
- ): Promise<RuntimeReadModelSessionView> {
- return this.readModel(projectionCache).getSessionView(sessionId);
+ private async getSessionView(sessionId: string): Promise<RuntimeReadModelSessionView> {
+ return this.readModel().getSessionView(sessionId);
}
- private readModel(
- projectionCache: RuntimeReadModelProjectionCache = this.deps.store,
- ): RuntimeReadModel {
+ private readModel(): RuntimeReadModel {
if (!this.deps.runStore || !this.deps.runtimeEventStore) {
throw new Error('RuntimeReadModel requires AgentRunStore and RuntimeEventStore');
}
return new RuntimeReadModel({
runtimeEventStore: this.deps.runtimeEventStore,
- projectionCache,
...(this.deps.canonicalPermissionOutcomes
? { canonicalPermissionOutcomes: this.deps.canonicalPermissionOutcomes }
: {}),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org