| import { produce } from 'immer'; |
|
|
| import { BugIndicatingError } from '#/_base/errors/errors'; |
| import { onUnexpectedError } from '#/_base/errors/unexpectedError'; |
| import { ILogService } from '#/_base/log/log'; |
| import { Service } from '#/_base/di/service'; |
| import { toDisposable, type IDisposable } from '#/_base/di/lifecycle'; |
| import { type CollectionView } from '#/_base/di/collection'; |
| import { LifecycleScope } from '#/app/scopes'; |
| import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; |
| import { AgentSpaceImpl, type AgentSpaceHost } from '#/agent/agentContext/agentSpace'; |
| import { IAgentBlobService } from '#/agent/blob/agentBlobService'; |
| import { IAgentScopeContext } from '#/agent/scopeContext/scopeContext'; |
| import { IAgentStateService } from '#/agent/state/agentState'; |
| import { |
| event2FromRecord, |
| type AgentDomainTrait, |
| type Event2, |
| type Event2Class, |
| } from '#/app/event/event2'; |
| import { IEventBus } from '#/app/event/eventBus'; |
| import type { ContentPart } from '#human/llm/message'; |
| import { OrderedHookSlot } from '#/hooks'; |
| import { IWireService } from '#/wire/wire'; |
| import { WireError, WireErrors } from '#/wire/errors'; |
| import { isHumanRecordType } from '#/wire/human'; |
| import { AGENT_SWITCHED_TYPE } from '#/wire/tree/index'; |
| import type { PartsTransformer } from '#/wire/record'; |
|
|
| import { |
| AgentModelContribution, |
| agentModelDefinitions, |
| type AgentModel, |
| type AgentModelDefinition, |
| } from './agentModel'; |
| import { IEventDispatcher, type DurableAgentRuntimeParticipant, type RestorePhase } from './eventDispatcher'; |
| import { StateError, StateErrors } from './errors'; |
| import { |
| expandedModelAppliers, |
| expandedRuntimeFolds, |
| type EventApplier, |
| type StateFold, |
| type FoldContext, |
| type ReplayableStateKey, |
| } from './state'; |
| import { |
| EventStateContribution, |
| foldEventStateContributions, |
| type EventStateContributionRecord, |
| type FoldedEventStateRegistry, |
| } from './stateContribution'; |
|
|
| const MAX_DRAIN = 100; |
|
|
| const UNREPORTED_WIRE_RECORD_TYPES: ReadonlySet<string> = new Set([ |
| 'staleGuard.recorded', |
| 'staleGuard.cleared', |
| AGENT_SWITCHED_TYPE, |
| 'context.undone', |
| ]); |
|
|
| export class CycleError extends StateError { |
| constructor(readonly depth: number, readonly eventTypes: readonly string[]) { |
| super( |
| StateErrors.codes.STATE_CYCLE, |
| `Event dispatch cascade exceeded MAX_DRAIN (${depth}); possible event cycle`, |
| { details: { depth, eventTypes: eventTypes.slice(0, 20) } }, |
| ); |
| this.name = 'CycleError'; |
| } |
| } |
|
|
| interface StateMeta { |
| checkpoints: unknown[]; |
| } |
|
|
| interface QueuedEvent { |
| readonly event: Event2<any>; |
| readonly resolve: () => void; |
| readonly reject: (error: unknown) => void; |
| } |
|
|
| interface PreparedFold { |
| readonly key: ReplayableStateKey<any>; |
| readonly meta: StateMeta; |
| readonly ctx: FoldContextImpl; |
| readonly next: any; |
| } |
|
|
| type ParticipantApplier = ( |
| state: any, |
| event: Event2<any>, |
| ctx: FoldContextImpl, |
| ) => unknown; |
|
|
| interface ParticipantAttachment { |
| readonly id: string; |
| readonly appliers: ReadonlyMap<Event2Class<any, any>, ParticipantApplier>; |
| readonly meta: StateMeta; |
| readonly undoable: boolean; |
| readonly initial: unknown; |
| readonly getState: () => any; |
| readonly commit: (state: any) => void; |
| } |
|
|
| interface PreparedParticipant { |
| readonly attachment: ParticipantAttachment; |
| readonly ctx: FoldContextImpl; |
| readonly next: any; |
| } |
|
|
| class FoldContextImpl implements FoldContext { |
| pendingCheckpoint = false; |
| pendingClear = false; |
| pendingUndo: number | undefined; |
|
|
| constructor( |
| private readonly owner: EventDispatcherService, |
| readonly silent: boolean, |
| ) {} |
|
|
| checkpoint(): void { |
| if (!this.silent) return; |
| this.pendingCheckpoint = true; |
| } |
|
|
| clearCheckpoints(): void { |
| if (!this.silent) return; |
| this.pendingClear = true; |
| } |
|
|
| undoToCheckpoint(count: number): void { |
| if (!this.silent) return; |
| this.pendingUndo = count; |
| } |
|
|
| emit(event: Event2<any>): void { |
| if (this.silent) return; |
| this.owner.enqueue(event); |
| } |
| } |
|
|
| function sanitizePendingUndo(ctx: FoldContextImpl, meta: StateMeta): void { |
| if ( |
| ctx.pendingUndo !== undefined && |
| (!Number.isSafeInteger(ctx.pendingUndo) || |
| ctx.pendingUndo <= 0 || |
| meta.checkpoints.length < ctx.pendingUndo) |
| ) { |
| ctx.pendingUndo = undefined; |
| } |
| } |
|
|
| export class EventDispatcherService extends Service implements IEventDispatcher { |
| declare readonly _serviceBrand: undefined; |
|
|
| readonly hooks: IEventDispatcher['hooks'] = { |
| onDidRestore: new OrderedHookSlot(), |
| }; |
|
|
| private readonly metas = new Map<ReplayableStateKey<any>, StateMeta>(); |
| private folded: FoldedEventStateRegistry; |
|
|
| private activeModelDefs = new Map<string, AgentModelDefinition<any, any>>(); |
| private readonly withdrawnModelIds = new Set<string>(); |
| private modelTargets = new Map<string, readonly AgentModelDefinition<any, any>[]>(); |
| private readonly modelAttachments = new Map< |
| AgentModelDefinition<any, any>, |
| ParticipantAttachment |
| >(); |
| private readonly participantTargets = new Map<string, ParticipantAttachment[]>(); |
| private readonly participantAttachments = new Map<string, ParticipantAttachment>(); |
|
|
| private readonly spaceHost: AgentSpaceHost = { |
| isActiveModelDefinition: (definition) => |
| this.activeModelDefs.get(definition.id) === definition, |
| registerModel: (definition, model) => this.registerModel(definition, model), |
| dispatchModelEvent: (event) => this.dispatch(event), |
| readLegacyState: (key) => this.agentState.get(key), |
| }; |
|
|
| restorePhase: RestorePhase = 'new'; |
| private dispatching = false; |
| private disposed = false; |
| private queue: QueuedEvent[] = []; |
| private drainDepth = 0; |
| private didRunRestoreHooks = false; |
| private lateAttachments: Array<{ |
| readonly participant: DurableAgentRuntimeParticipant; |
| readonly resolve: (disposable: IDisposable) => void; |
| readonly reject: (error: unknown) => void; |
| }> = []; |
|
|
| constructor( |
| @IWireService private readonly wire: IWireService, |
| @IEventBus private readonly eventBus: IEventBus, |
| @IAgentScopeContext private readonly agentScope: IAgentScopeContext | undefined, |
| @IAgentBlobService private readonly blobService: IAgentBlobService, |
| @IAgentStateService private readonly agentState: IAgentStateService, |
| @ILogService private readonly logger: ILogService, |
| @EventStateContribution view: CollectionView<EventStateContributionRecord>, |
| @AgentModelContribution modelView: CollectionView<AgentModelDefinition<any, any>>, |
| ) { |
| super(); |
| this.folded = this.foldContributions(view); |
| this._register( |
| view.onDidChange(() => { |
| this.folded = this.foldContributions(view); |
| }), |
| ); |
| this._register( |
| this.agentState.onDidContributeReplayable((key) => { |
| if (this.restorePhase !== 'new') { |
| throw new BugIndicatingError( |
| `Replayable state '${key.name}' contributed while the event dispatcher is in phase '${this.restorePhase}'; replayable state owners must contribute before restore`, |
| ); |
| } |
| this.folded = this.foldContributions(view); |
| }), |
| ); |
| this._register( |
| this.agentState.onDidWithdrawReplayable((key) => { |
| this.metas.delete(key); |
| this.folded = this.foldContributions(view); |
| }), |
| ); |
| this.refoldModels(modelView.items); |
| this._register( |
| modelView.onDidChange(({ added, removed }) => { |
| for (const definition of removed) { |
| this.withdrawnModelIds.add(definition.id); |
| const attachment = this.modelAttachments.get(definition); |
| if (attachment !== undefined) { |
| this.modelAttachments.delete(definition); |
| this.detachParticipant(attachment); |
| this.space()?.retireModel(definition); |
| } |
| } |
| for (const definition of added) { |
| this.withdrawnModelIds.delete(definition.id); |
| } |
| this.refoldModels(modelView.items); |
| this.materializeUndoableModels(); |
| }), |
| ); |
| this.space()?._attachHost(this.spaceHost); |
| this.materializeUndoableModels(); |
| } |
|
|
| private space(): AgentSpaceImpl | undefined { |
| const space = this.agentScope?.agentContext.space; |
| return space instanceof AgentSpaceImpl ? space : undefined; |
| } |
|
|
| private foldContributions( |
| view: CollectionView<EventStateContributionRecord>, |
| ): FoldedEventStateRegistry { |
| return foldEventStateContributions(view.items, this.agentState.replayableKeys()); |
| } |
|
|
| attach(participant: DurableAgentRuntimeParticipant): IDisposable { |
| if (this.restorePhase !== 'new') { |
| throw new BugIndicatingError( |
| `Agent runtime participant '${participant.id}' attached while the event dispatcher is in phase '${this.restorePhase}'; durable runtime owners must attach before restore`, |
| ); |
| } |
| const attachment = this.buildParticipantAttachment(participant); |
| this.attachParticipant(attachment); |
| return toDisposable(() => { this.detachParticipant(attachment); }); |
| } |
|
|
| async attachLate(participant: DurableAgentRuntimeParticipant): Promise<IDisposable> { |
| if (this.restorePhase === 'restoring') { |
| return new Promise<IDisposable>((resolve, reject) => { |
| this.lateAttachments.push({ participant, resolve, reject }); |
| }); |
| } |
| if (this.restorePhase !== 'ready') { |
| throw new BugIndicatingError( |
| `Agent runtime participant '${participant.id}' late-attached while the event dispatcher is in phase '${this.restorePhase}'; late attach requires a restored dispatcher`, |
| ); |
| } |
| return this.attachLateNow(participant); |
| } |
|
|
| private async attachLateNow(participant: DurableAgentRuntimeParticipant): Promise<IDisposable> { |
| if (this.disposed) { |
| throw new Error(`Agent runtime participant '${participant.id}' late-attached to a disposed event dispatcher`); |
| } |
| const attachment = this.buildParticipantAttachment(participant); |
| this.dispatching = true; |
| try { |
| await this.wire.flush(); |
| const stream = participant.undoable |
| ? this.wire.readRestorable() |
| : this.wire.readJournal(); |
| for await (const record of stream) { |
| if (record.type === 'metadata') continue; |
| const cls = this.folded.events.get(record.type); |
| if (cls === undefined) continue; |
| let eventRecord = record; |
| if (cls.agentDomain) { |
| if (this.agentScope === undefined) continue; |
| const recordAgentId = record['agentId']; |
| if (recordAgentId === undefined) eventRecord = { ...record, agentId: this.agentScope.agentId }; |
| else if (recordAgentId !== this.agentScope.agentId) continue; |
| } |
| const event = event2FromRecord(cls, eventRecord); |
| if (event === undefined) continue; |
| const applier = attachment.appliers.get(event.constructor as Event2Class); |
| if (applier === undefined) continue; |
| const ctx = new FoldContextImpl(this, true); |
| const next = produce( |
| attachment.getState(), |
| (draft: any) => applier(draft, event, ctx), |
| ); |
| if (ctx.pendingUndo !== undefined && next !== attachment.getState()) { |
| throw new BugIndicatingError( |
| `Fold of event '${event.type}' on durable participant '${attachment.id}' both mutates and undoes to a checkpoint`, |
| ); |
| } |
| sanitizePendingUndo(ctx, attachment.meta); |
| this.commitParticipant(attachment, ctx, next); |
| } |
| this.attachParticipant(attachment); |
| this.drainQueue(); |
| } catch (error) { |
| for (const entry of this.queue.splice(0)) entry.reject(error); |
| throw error; |
| } finally { |
| this.queue.length = 0; |
| this.dispatching = false; |
| this.drainDepth = 0; |
| } |
| return toDisposable(() => { this.detachParticipant(attachment); }); |
| } |
|
|
| private buildParticipantAttachment( |
| participant: DurableAgentRuntimeParticipant, |
| ): ParticipantAttachment { |
| const base = new Map<Event2Class<any, any>, StateFold<any, any>>(); |
| for (const cls of participant.events) base.set(cls, participant.transition); |
| const folds = expandedRuntimeFolds(participant.id, participant.undoable, base); |
| const appliers = new Map<Event2Class<any, any>, ParticipantApplier>(); |
| for (const [cls, fold] of folds) { |
| appliers.set(cls, (state, event, ctx) => fold(state, event, ctx)); |
| } |
| return { |
| id: participant.id, |
| appliers, |
| meta: { checkpoints: [] }, |
| undoable: participant.undoable, |
| initial: participant.getState(), |
| getState: () => participant.getState(), |
| commit: (state) => { participant.commit(state); }, |
| }; |
| } |
|
|
| private attachParticipant(attachment: ParticipantAttachment): void { |
| if (this.participantAttachments.has(attachment.id)) { |
| throw new BugIndicatingError(`Durable participant '${attachment.id}' is already attached`); |
| } |
| this.participantAttachments.set(attachment.id, attachment); |
| for (const cls of attachment.appliers.keys()) { |
| const list = this.participantTargets.get(cls.type) ?? []; |
| list.push(attachment); |
| this.participantTargets.set(cls.type, list); |
| } |
| } |
|
|
| private detachParticipant(attachment: ParticipantAttachment): void { |
| if (this.participantAttachments.get(attachment.id) !== attachment) return; |
| this.participantAttachments.delete(attachment.id); |
| for (const cls of attachment.appliers.keys()) { |
| const list = this.participantTargets.get(cls.type); |
| if (list === undefined) continue; |
| const next = list.filter((candidate) => candidate !== attachment); |
| if (next.length === 0) this.participantTargets.delete(cls.type); |
| else this.participantTargets.set(cls.type, next); |
| } |
| } |
|
|
| private refoldModels(records: readonly AgentModelDefinition<any, any>[]): void { |
| const defs = new Map<string, AgentModelDefinition<any, any>>(); |
| for (const definition of agentModelDefinitions()) { |
| if (!this.withdrawnModelIds.has(definition.id)) defs.set(definition.id, definition); |
| } |
| for (const definition of records) defs.set(definition.id, definition); |
| this.activeModelDefs = defs; |
| this.rebuildModelTargets(); |
| } |
|
|
| private rebuildModelTargets(): void { |
| const targets = new Map<string, AgentModelDefinition<any, any>[]>(); |
| const add = (type: string, definition: AgentModelDefinition<any, any>): void => { |
| const list = targets.get(type); |
| if (list === undefined) { |
| targets.set(type, [definition]); |
| return; |
| } |
| if (!list.includes(definition)) list.push(definition); |
| }; |
| const domainOwners = new Map<string, AgentModelDefinition<any, any>>(); |
| for (const definition of this.activeModelDefs.values()) { |
| for (const cls of definition.events) { |
| const owner = domainOwners.get(cls.type); |
| if (owner !== undefined && owner !== definition) { |
| throw new BugIndicatingError( |
| `Event '${cls.type}' is applied by both agent models '${owner.id}' and '${definition.id}'`, |
| ); |
| } |
| domainOwners.set(cls.type, definition); |
| add(cls.type, definition); |
| } |
| } |
| for (const [definition, attachment] of this.modelAttachments) { |
| if (this.activeModelDefs.get(definition.id) !== definition) continue; |
| for (const cls of attachment.appliers.keys()) add(cls.type, definition); |
| } |
| this.modelTargets = targets; |
| } |
|
|
| private materializeUndoableModels(): void { |
| const space = this.space(); |
| if (space === undefined) return; |
| for (const definition of this.activeModelDefs.values()) { |
| if (!definition.undoable || this.modelAttachments.has(definition)) continue; |
| space.ensureModel(definition); |
| } |
| } |
|
|
| private registerModel( |
| definition: AgentModelDefinition<any, any>, |
| model: AgentModel<any>, |
| ): void { |
| if (this.modelAttachments.has(definition)) return; |
| const domainAppliers = new Map<Event2Class<any, any>, EventApplier>(); |
| for (const [cls, applier] of model._appliersTable()) { |
| domainAppliers.set(cls, (event) => applier.call(model, event)); |
| } |
| const customUndo = |
| model.onUndo === undefined ? undefined : (count: number): void => model.onUndo!(count); |
| const expanded = expandedModelAppliers( |
| definition.id, |
| definition.undoable, |
| domainAppliers, |
| customUndo, |
| ); |
| const appliers = new Map<Event2Class<any, any>, ParticipantApplier>(); |
| for (const [cls, applier] of expanded) { |
| appliers.set(cls, (state, event, ctx) => { |
| model._enterWindow(state, ctx); |
| let windowResult: ReturnType<AgentModel<any>['_exitWindow']>; |
| try { |
| applier(event, ctx); |
| } finally { |
| windowResult = model._exitWindow(); |
| } |
| return windowResult.replaced ? windowResult.replacement : undefined; |
| }); |
| } |
| const attachment: ParticipantAttachment = { |
| id: definition.id, |
| appliers, |
| meta: { checkpoints: [] }, |
| undoable: definition.undoable, |
| initial: model._state(), |
| getState: () => model._state(), |
| commit: (state) => { model._commitState(state); }, |
| }; |
| this.attachParticipant(attachment); |
| this.modelAttachments.set(definition, attachment); |
| this.rebuildModelTargets(); |
| } |
|
|
| private materializeModel(definition: AgentModelDefinition<any, any>): ParticipantAttachment { |
| const space = this.space(); |
| if (space === undefined) { |
| throw new BugIndicatingError( |
| `Agent model '${definition.id}' cannot materialize without an agent space`, |
| ); |
| } |
| space.ensureModel(definition); |
| const attachment = this.modelAttachments.get(definition); |
| if (attachment === undefined) { |
| throw new BugIndicatingError(`Agent model '${definition.id}' failed to attach`); |
| } |
| return attachment; |
| } |
|
|
| dispatch(event: Event2<any>): Promise<void> { |
| const cls = event.constructor as Event2Class; |
| if ( |
| cls.agentDomain && |
| (this.agentScope === undefined || |
| (event as Event2<any> & AgentDomainTrait).agentId !== this.agentScope.agentId) |
| ) { |
| return Promise.reject( |
| new Error(`Agent event '${event.type}' does not match dispatcher lifecycle context`), |
| ); |
| } |
| if (this.dispatching) { |
| return new Promise<void>((resolve, reject) => { |
| this.queue.push({ event, resolve, reject }); |
| }); |
| } |
| this.dispatching = true; |
| try { |
| this.runDispatch(event); |
| this.drainQueue(); |
| return Promise.resolve(); |
| } catch (error) { |
| for (const entry of this.queue.splice(0)) { |
| entry.reject(error); |
| } |
| return Promise.reject(error); |
| } finally { |
| this.queue.length = 0; |
| this.dispatching = false; |
| this.drainDepth = 0; |
| } |
| } |
|
|
| private drainQueue(): void { |
| while (this.queue.length > 0) { |
| if (++this.drainDepth > MAX_DRAIN) { |
| throw new CycleError( |
| this.drainDepth, |
| this.queue.map((entry) => entry.event.type), |
| ); |
| } |
| const entry = this.queue.shift()!; |
| try { |
| this.runDispatch(entry.event); |
| entry.resolve(); |
| } catch (error) { |
| entry.reject(error); |
| throw error; |
| } |
| } |
| } |
|
|
| enqueue(event: Event2<any>): void { |
| this.queue.push({ |
| event, |
| resolve: () => {}, |
| reject: (error: unknown) => onUnexpectedError(error), |
| }); |
| } |
|
|
| private runDispatch(event: Event2<any>): void { |
| this.executeEvent(event, false); |
| } |
|
|
| private executeEvent(event: Event2<any>, silent: boolean, replayUndoable?: boolean): void { |
| const folds = this.folded.folds.get(event.type); |
| const prepared: PreparedFold[] = []; |
| if (folds !== undefined) { |
| for (const { key, fold } of folds) { |
| if ( |
| replayUndoable !== undefined && |
| (key.replayable.undoable !== undefined) !== replayUndoable |
| ) { |
| continue; |
| } |
| const meta = this.ensureMeta(key); |
| const ctx = new FoldContextImpl(this, silent); |
| const next = produce( |
| this.agentState.get(key), |
| (draft: any) => fold(draft, event, ctx), |
| ); |
| if (ctx.pendingUndo !== undefined && next !== this.agentState.get(key)) { |
| throw new BugIndicatingError( |
| `Fold of event '${event.type}' on state '${key.name}' both mutates and undoes to a checkpoint`, |
| ); |
| } |
| sanitizePendingUndo(ctx, meta); |
| prepared.push({ key, meta, ctx, next }); |
| } |
| } |
| const modelTargets = this.modelTargets.get(event.type); |
| if (modelTargets !== undefined) { |
| for (const definition of modelTargets) { |
| if (replayUndoable !== undefined && definition.undoable !== replayUndoable) continue; |
| if (!this.modelAttachments.has(definition)) this.materializeModel(definition); |
| } |
| } |
| const participantTargets = this.participantTargets.get(event.type); |
| const preparedParticipants: PreparedParticipant[] = []; |
| if (participantTargets !== undefined) { |
| for (const attachment of participantTargets) { |
| if (replayUndoable !== undefined && attachment.undoable !== replayUndoable) continue; |
| const applier = attachment.appliers.get(event.constructor as Event2Class); |
| if (applier === undefined) continue; |
| const ctx = new FoldContextImpl(this, silent); |
| const next = produce( |
| attachment.getState(), |
| (draft: any) => applier(draft, event, ctx), |
| ); |
| if (ctx.pendingUndo !== undefined && next !== attachment.getState()) { |
| throw new BugIndicatingError( |
| `Fold of event '${event.type}' on durable participant '${attachment.id}' both mutates and undoes to a checkpoint`, |
| ); |
| } |
| sanitizePendingUndo(ctx, attachment.meta); |
| preparedParticipants.push({ attachment, ctx, next }); |
| } |
| } |
| for (const p of prepared) { |
| this.commit(p.key, p.meta, p.ctx, p.next); |
| } |
| for (const p of preparedParticipants) { |
| this.commitParticipant(p.attachment, p.ctx, p.next); |
| } |
| if (silent) return; |
| const cls = event.constructor as Event2Class; |
| if (cls.durable) { |
| const dehydrator = folds?.find(({ key }) => key.replayable.blobs !== undefined)?.key |
| .replayable.blobs?.dehydrate; |
| this.wire.appendRecord(event.serialize(), dehydrator); |
| } |
| if (cls.observable && !this.disposed) { |
| this.eventBus.publish(event, this.agentScope?.agentContext); |
| } |
| } |
|
|
| override dispose(): void { |
| this.disposed = true; |
| const pending = this.lateAttachments.splice(0); |
| if (pending.length > 0) { |
| const error = new Error('Event dispatcher disposed while a late attach was pending'); |
| for (const entry of pending) entry.reject(error); |
| } |
| this.space()?._detachHost(this.spaceHost); |
| super.dispose(); |
| } |
|
|
| private commit( |
| key: ReplayableStateKey<any>, |
| meta: StateMeta, |
| ctx: FoldContextImpl, |
| next: any, |
| ): void { |
| if (ctx.pendingUndo !== undefined) { |
| const targetIndex = meta.checkpoints.length - ctx.pendingUndo; |
| const snapshot = meta.checkpoints[targetIndex]!; |
| this.agentState.set(key, snapshot); |
| meta.checkpoints.length = targetIndex; |
| return; |
| } |
| this.agentState.set(key, next); |
| if (ctx.pendingClear) { |
| meta.checkpoints.length = 0; |
| } |
| if (ctx.pendingCheckpoint) { |
| meta.checkpoints.push(next); |
| } |
| } |
|
|
| private commitParticipant( |
| attachment: ParticipantAttachment, |
| ctx: FoldContextImpl, |
| next: any, |
| ): void { |
| const meta = attachment.meta; |
| if (ctx.pendingUndo !== undefined) { |
| const targetIndex = meta.checkpoints.length - ctx.pendingUndo; |
| const snapshot = meta.checkpoints[targetIndex]!; |
| attachment.commit(snapshot); |
| meta.checkpoints.length = targetIndex; |
| return; |
| } |
| attachment.commit(next); |
| if (ctx.pendingClear) { |
| meta.checkpoints.length = 0; |
| } |
| if (ctx.pendingCheckpoint) { |
| meta.checkpoints.push(next); |
| } |
| } |
|
|
| private ensureMeta(key: ReplayableStateKey<any>): StateMeta { |
| let meta = this.metas.get(key); |
| if (meta === undefined) { |
| meta = { checkpoints: [] }; |
| this.metas.set(key, meta); |
| } |
| return meta; |
| } |
|
|
| async restore(): Promise<void> { |
| if (this.restorePhase === 'restoring') { |
| throw new BugIndicatingError( |
| `Agent state restore called while phase is ${this.restorePhase}`, |
| ); |
| } |
| const rerun = this.restorePhase !== 'new'; |
| this.restorePhase = 'restoring'; |
| if (rerun) this.dispatching = true; |
| try { |
| if (rerun) { |
| await this.wire.flush(); |
| this.resetReplayState(); |
| } |
| await this.replayRecords(true); |
| await this.replayRecords(false); |
| await this.rehydrateStates(); |
| this.restorePhase = 'ready'; |
| if (!this.didRunRestoreHooks) { |
| await this.hooks.onDidRestore.run({}); |
| this.didRunRestoreHooks = true; |
| } |
| if (rerun) { |
| this.drainQueue(); |
| } |
| await this.drainLateAttachments(); |
| } catch (error) { |
| this.restorePhase = 'failed'; |
| for (const pending of this.lateAttachments.splice(0)) pending.reject(error); |
| if (rerun) { |
| for (const entry of this.queue.splice(0)) entry.reject(error); |
| } |
| throw error; |
| } finally { |
| if (rerun) { |
| this.queue.length = 0; |
| this.dispatching = false; |
| this.drainDepth = 0; |
| } |
| } |
| } |
|
|
| private async drainLateAttachments(): Promise<void> { |
| for (const pending of this.lateAttachments.splice(0)) { |
| try { |
| pending.resolve(await this.attachLateNow(pending.participant)); |
| } catch (error) { |
| pending.reject(error); |
| } |
| } |
| } |
|
|
| private resetReplayState(): void { |
| for (const key of this.agentState.replayableKeys()) { |
| this.agentState.set(key, key.initial()); |
| } |
| for (const attachment of this.participantAttachments.values()) { |
| attachment.commit(attachment.initial); |
| attachment.meta.checkpoints.length = 0; |
| } |
| this.metas.clear(); |
| } |
|
|
| private async replayRecords(undoable: boolean): Promise<void> { |
| const stream = undoable ? this.wire.readRestorable() : this.wire.readJournal(); |
| let recordIndex = 0; |
| for await (const record of stream) { |
| if (record.type === 'metadata') continue; |
| const cls = this.folded.events.get(record.type); |
| if (cls === undefined) { |
| if ( |
| !undoable && |
| !UNREPORTED_WIRE_RECORD_TYPES.has(record.type) && |
| !isHumanRecordType(record.type) |
| ) { |
| this.reportSkippedRecord(record.type, recordIndex, false); |
| } |
| recordIndex++; |
| continue; |
| } |
| let eventRecord = record; |
| if (cls.agentDomain) { |
| if (this.agentScope === undefined) { |
| if (!undoable) this.reportSkippedRecord(record.type, recordIndex, true); |
| recordIndex++; |
| continue; |
| } |
| const recordAgentId = record['agentId']; |
| if (recordAgentId === undefined) { |
| eventRecord = { ...record, agentId: this.agentScope.agentId }; |
| } else if (recordAgentId !== this.agentScope.agentId) { |
| if (!undoable) this.reportSkippedRecord(record.type, recordIndex, true); |
| recordIndex++; |
| continue; |
| } |
| } |
| const event = event2FromRecord(cls, eventRecord); |
| if (event === undefined) { |
| if (!undoable) this.reportSkippedRecord(record.type, recordIndex, true); |
| recordIndex++; |
| continue; |
| } |
| this.executeEvent(event, true, undoable); |
| recordIndex++; |
| } |
| } |
|
|
| private reportSkippedRecord(type: string, index: number, malformed: boolean): void { |
| const message = malformed |
| ? `Malformed wire record type '${type}' skipped during restore` |
| : `Unknown wire record type '${type}' skipped during restore`; |
| if (malformed) { |
| onUnexpectedError( |
| new WireError(WireErrors.codes.WIRE_UNKNOWN_RECORD, message, { details: { type, index } }), |
| ); |
| return; |
| } |
| this.logger.warn(message, { code: WireErrors.codes.WIRE_UNKNOWN_RECORD, type, index }); |
| } |
|
|
| private async rehydrateStates(): Promise<void> { |
| const transform: PartsTransformer = (parts) => |
| this.blobService.loadParts(parts as readonly ContentPart[]) as Promise<readonly unknown[]>; |
| for (const key of this.folded.states) { |
| const codec = key.replayable.blobs; |
| if (codec?.rehydrate === undefined) continue; |
| this.agentState.set(key, Object.freeze(await codec.rehydrate(this.agentState.get(key), transform))); |
| } |
| } |
|
|
| async flush(): Promise<void> { |
| await this.wire.flush(); |
| } |
| } |
|
|
| registerScopedService( |
| LifecycleScope.Agent, |
| IEventDispatcher, |
| EventDispatcherService, |
| ScopeActivation.OnScopeCreated, |
| 'state', |
| ); |
|
|
