refactor(agent-core-v2): merge event sink into IAgentRecordService

- add IAgentRecordService: append(record) fans out to durable persistence,
  live broadcast, and replay capture via per-type facets
  (toLive/toReplay/resume/blobs); signal(event) emits live-only events
- remove IAgentEventSinkService; fold live emission into IAgentRecordService
- migrate all agent domains to record.append/signal/define
- server-v2 subscribes to agent events via IAgentRecordService
- delete dead createLoopEventDispatcher
- AgentRecordMap extends WireRecordMap so record types stay on WireRecordMap
This commit is contained in:
haozhe.yang 2026-07-01 22:38:25 +08:00
parent 4f81640285
commit fff2babe42
43 changed files with 1229 additions and 454 deletions

View file

@ -63,7 +63,7 @@ package "Session scope (per session)" #EAFAF1 {
package "Agent scope (per agent)" #FDF5E6 {
rectangle "<b>wireRecord</b>\n<size:9><i>Agent</i></size>\n IAgentWireRecordService (event hub)" as wireRecord #FDEBD0
rectangle "<b>eventSink</b>\n<size:9><i>Agent</i></size>\n IAgentEventSinkService" as eventSink #FDEBD0
rectangle "<b>record</b>\n<size:9><i>Agent</i></size>\n IAgentRecordService" as record #FDEBD0
rectangle "<b>blobStore</b>\n<size:9><i>Agent</i></size>\n IAgentBlobStoreService" as blobStore #FDEBD0
rectangle "<b>contextMemory</b>\n<size:9><i>Agent</i></size>\n IAgentContextMemoryService" as contextMemory #FDEBD0
rectangle "<b>contextProjector</b>\n<size:9><i>Agent</i></size>\n IAgentContextProjectorService" as contextProjector #FDEBD0
@ -109,7 +109,7 @@ package "Agent scope (per agent)" #FDF5E6 {
' ---- DI injection (solid) ----
gateway --> session_lifecycle #34495E
gateway --> log #34495E
gateway --> eventSink #34495E
gateway --> record #34495E
sessionIndex --> bootstrap #34495E
sessionIndex --> storage #34495E
session_lifecycle --> bootstrap #34495E
@ -165,19 +165,19 @@ wireRecord --> bootstrap #34495E
wireRecord --> storage #34495E
blobStore --> storage #34495E
filestore --> storage #34495E
eventSink --> wireRecord #34495E
record --> wireRecord #34495E
contextMemory --> wireRecord #34495E
contextMemory --> replayBuilder #34495E
contextInjector --> contextMemory #34495E
contextInjector --> turn #34495E
contextInjector --> systemReminder #34495E
contextSize --> contextMemory #34495E
contextSize --> eventSink #34495E
contextSize --> record #34495E
contextSize --> wireRecord #34495E
systemReminder --> contextMemory #34495E
replayBuilder --> wireRecord #34495E
profile --> wireRecord #34495E
profile --> eventSink #34495E
profile --> record #34495E
profile --> replayBuilder #34495E
profile --> telemetry #34495E
profile --> config #34495E
@ -186,9 +186,9 @@ profile --> chatProvider #34495E
prompt --> contextMemory #34495E
prompt --> turn #34495E
prompt --> wireRecord #34495E
prompt --> eventSink #34495E
prompt --> record #34495E
turn --> loop #34495E
turn --> eventSink #34495E
turn --> record #34495E
turn --> wireRecord #34495E
turn --> contextMemory #34495E
turn --> externalHooks #34495E
@ -197,7 +197,7 @@ loop --> contextMemory #34495E
loop --> contextProjector #34495E
loop --> contextSize #34495E
loop --> llmRequester #34495E
loop --> eventSink #34495E
loop --> record #34495E
loop --> toolRegistry #34495E
loop --> toolExecutor #34495E
loop --> profile #34495E
@ -226,7 +226,7 @@ permissionGate --> externalHooks #34495E
permissionGate --> telemetry #34495E
permissionGate --> toolExecutor #34495E
permissionMode --> wireRecord #34495E
permissionMode --> eventSink #34495E
permissionMode --> record #34495E
permissionMode --> replayBuilder #34495E
permissionMode --> contextInjector #34495E
permissionRules --> wireRecord #34495E
@ -234,7 +234,7 @@ permissionRules --> replayBuilder #34495E
permissionRules --> config #34495E
plan --> contextMemory #34495E
plan --> wireRecord #34495E
plan --> eventSink #34495E
plan --> record #34495E
plan --> agentFs #34495E
plan --> profile #34495E
plan --> replayBuilder #34495E
@ -242,13 +242,13 @@ plan --> toolRegistry #34495E
plan --> contextInjector #34495E
plan --> telemetry #34495E
goal --> wireRecord #34495E
goal --> eventSink #34495E
goal --> record #34495E
goal --> systemReminder #34495E
goal --> replayBuilder #34495E
goal --> telemetry #34495E
goal --> contextInjector #34495E
skill --> prompt #34495E
skill --> eventSink #34495E
skill --> record #34495E
skill --> wireRecord #34495E
skill --> telemetry #34495E
skill --> sessionSkillCatalog #34495E
@ -259,7 +259,7 @@ questionTools --> telemetry #34495E
userTool --> toolRegistry #34495E
userTool --> profile #34495E
userTool --> wireRecord #34495E
background --> eventSink #34495E
background --> record #34495E
background --> wireRecord #34495E
background --> telemetry #34495E
background --> prompt #34495E
@ -269,7 +269,7 @@ background --> config #34495E
background --> storage #34495E
background --> session_context #34495E
cron --> prompt #34495E
cron --> eventSink #34495E
cron --> record #34495E
cron --> wireRecord #34495E
cron --> turn #34495E
cron --> telemetry #34495E
@ -277,11 +277,11 @@ cron --> toolRegistry #34495E
cron --> config #34495E
cron --> storage #34495E
swarm --> wireRecord #34495E
swarm --> eventSink #34495E
swarm --> record #34495E
swarm --> systemReminder #34495E
swarm --> agentTool #34495E
mcp --> toolRegistry #34495E
mcp --> eventSink #34495E
mcp --> record #34495E
mcp --> toolExecutor #34495E
fullCompaction --> contextMemory #34495E
fullCompaction --> contextProjector #34495E
@ -291,7 +291,7 @@ fullCompaction --> profile #34495E
fullCompaction --> toolStore #34495E
fullCompaction --> telemetry #34495E
fullCompaction --> wireRecord #34495E
fullCompaction --> eventSink #34495E
fullCompaction --> record #34495E
fullCompaction --> replayBuilder #34495E
fullCompaction --> externalHooks #34495E
fullCompaction --> turn #34495E
@ -310,7 +310,7 @@ todoList --> toolStore #34495E
todoList --> toolRegistry #34495E
todoList --> contextInjector #34495E
usage --> wireRecord #34495E
usage --> eventSink #34495E
usage --> record #34495E
rpc --> prompt #34495E
rpc --> turn #34495E
rpc --> profile #34495E

View file

@ -56,7 +56,7 @@ import { IAgentContextMemoryService } from '#/agent/contextMemory';
import { IAgentContextProjectorService } from '#/agent/contextProjector';
import { IAgentContextSizeService } from '#/agent/contextSize';
import { AgentCronService } from '#/agent/cron';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import { AgentExternalHooksService, IAgentExternalHooksService } from '#/agent/externalHooks';
import { IFlagService } from '#/app/flag';
import { IAgentLLMRequesterService } from '#/agent/llmRequester';
@ -186,7 +186,7 @@ describe('config slice (every section owner against one shared registry)', () =>
// Collaborators declared but not touched during construction — empty
// stubs keep the container strict-clean (no "unknown service" warnings).
reg.definePartialInstance(IAgentEventSinkService, {});
reg.definePartialInstance(IAgentRecordService, {});
reg.definePartialInstance(ITelemetryService, { track: () => {} });
reg.definePartialInstance(IAgentPromptService, {});
reg.definePartialInstance(IAtomicDocumentStore, {});

View file

@ -81,7 +81,6 @@ const DOMAIN_LAYER = new Map([
['model', 2],
['session-index', 2],
['sessionStore', 2],
['eventSink', 2],
// L3 — registries & capabilities
['tool', 3],
['skill', 3],
@ -136,6 +135,9 @@ const DOMAIN_LAYER = new Map([
['mcp', 5],
['cron', 5],
['agentTool', 5],
// `btw` forks a single side-question sub-agent via `agent-lifecycle`, mirroring
// the `agentTool` shape (Agent-scope, spawns one child) — same layer.
['btw', 5],
// L6 — coordination
['agent-lifecycle', 6],
['session-lifecycle', 6],
@ -251,6 +253,7 @@ const ALLOWED_EXCEPTIONS = new Set([
'skill>prompt',
'swarm>agentTool',
'swarm>session-metadata',
'btw>agent-lifecycle',
'agentTool>agent-lifecycle',
'agentTool>session-metadata',
'toolExecutor>loop',

View file

@ -7,7 +7,7 @@
* - `ctor` constructor DI (`@IToken` param decorators)
* - `accessor` runtime lookups (`<expr>.get(IToken)`)
* - `publish`/`subscribe` `IEventService` usage from a class field
* - `emit`/`on` `IAgentEventSinkService` usage from a class field
* - `signal`/`append`/`on` `IAgentRecordService` usage from a class field
*
* Deliberately parse-only (no type checker) so the whole tree runs in ~1s.
* We rely on the codebase convention that constructor DI params carry an
@ -40,12 +40,13 @@ export const REPO_ROOT = resolve(PKG_ROOT, '..', '..');
export const SRC_ROOT = join(PKG_ROOT, 'src');
export const SNAPSHOT_PATH = join(PKG_ROOT, '.local', 'dep-graph.json');
const EVENT_BUS_TOKENS = new Set(['IEventService', 'IAgentEventSinkService']);
const EVENT_BUS_TOKENS = new Set(['IEventService', 'IAgentRecordService']);
const EVENT_METHOD_KIND: Record<string, EdgeKind> = {
publish: 'publish',
subscribe: 'subscribe',
emit: 'emit',
append: 'emit',
signal: 'emit',
on: 'on',
};
@ -160,9 +161,36 @@ function pushEdge(
token: string,
kind: EdgeKind,
ref: EdgeRef,
/**
* When set, resolve `token` from this scope instead of the source service's
* scope. Used for `<handle>.accessor.get(IX)` where the handle is statically
* known to belong to an inner scope (e.g. an `IAgentScopeHandle`), so the
* lookup resolves against that inner scope rather than the source's.
*/
overrideScope?: ServiceScope,
): void {
const target = resolveFromScope(acc.bindings, token, source.scope);
const toId = target ? target.id : `unresolved::${token}`;
const target = resolveFromScope(acc.bindings, token, overrideScope ?? source.scope);
// Classify a miss: the token is either unknown everywhere (genuinely
// unresolved) or registered at a scope the resolution scope can't see (a
// scope mismatch). The two render differently in the viewer.
let toId: string;
let extra: Pick<Edge, 'unresolved' | 'scopeMismatch' | 'actualScope'>;
if (target) {
toId = target.id;
extra = {};
} else {
const scopeMap = acc.bindings.get(token);
const actualScope = scopeMap ? innermostScope(scopeMap) : undefined;
if (actualScope !== undefined) {
toId = `scopeMismatch::${token}`;
extra = { scopeMismatch: true as const, actualScope };
} else {
toId = `unresolved::${token}`;
extra = { unresolved: true as const };
}
}
const key = edgeKey(fromId, toId, kind);
const existing = acc.edges.get(key);
if (existing) {
@ -177,10 +205,23 @@ function pushEdge(
token,
kind,
refs: [ref],
...(target ? {} : { unresolved: true as const }),
...extra,
};
acc.edges.set(key, edge);
if (!target) acc.unknownRefs.add(token);
if (extra.unresolved) acc.unknownRefs.add(token);
}
function innermostScope(scopeMap: Map<ServiceScope, ServiceNode>): ServiceScope | undefined {
let best: ServiceScope | undefined;
let bestLevel = -1;
for (const s of scopeMap.keys()) {
const lvl = SCOPE_LEVEL[s];
if (lvl > bestLevel) {
bestLevel = lvl;
best = s;
}
}
return best;
}
function sameRef(a: EdgeRef, b: EdgeRef): boolean {
@ -438,6 +479,251 @@ function chainedMethodName(getCall: CallExpression): string | undefined {
return pae.getName();
}
/**
* Map scope-typed handle aliases (and the generic `IScopeHandle<LifecycleScope.X>`
* form) to their scope. Used to resolve `<handle>.accessor.get(IX)` against the
* handle's real scope rather than the source service's scope.
*/
const HANDLE_ALIAS_SCOPE: Record<string, ServiceScope> = {
IAppScopeHandle: 'App',
ISessionScopeHandle: 'Session',
IAgentScopeHandle: 'Agent',
};
const FUNCTION_LIKE_KINDS = new Set<SyntaxKind>([
SyntaxKind.MethodDeclaration,
SyntaxKind.FunctionDeclaration,
SyntaxKind.ArrowFunction,
SyntaxKind.FunctionExpression,
SyntaxKind.Constructor,
SyntaxKind.GetAccessor,
SyntaxKind.SetAccessor,
]);
/** Strip `Promise<...>`, `| undefined` / `| null`, array brackets, `readonly`. */
function stripTypeWrappers(text: string): string {
let t = text.trim();
t = t.replace(/\s*\|\s*(undefined|null)\s*/g, '').trim();
const promise = /^Promise\s*<\s*(.+?)\s*>$/.exec(t);
if (promise) t = promise[1].trim();
t = t.replace(/\[\]\s*$/, '').trim();
t = t.replace(/^readonly\s+/, '').trim();
return t;
}
function handleScopeFromTypeText(text: string | undefined): ServiceScope | undefined {
if (text === undefined) return undefined;
const t = stripTypeWrappers(text);
const alias = HANDLE_ALIAS_SCOPE[t];
if (alias !== undefined) return alias;
const generic = /^IScopeHandle\s*<\s*LifecycleScope\.(App|Session|Agent)\s*>$/.exec(t);
if (generic) return generic[1] as ServiceScope;
return undefined;
}
/** Nearest function-like ancestor (method / function / arrow / ctor / accessor). */
function enclosingFunction(node: Node): Node | undefined {
let cur: Node | undefined = node.getParent();
while (cur) {
if (FUNCTION_LIKE_KINDS.has(cur.getKind())) return cur;
cur = cur.getParent();
}
return undefined;
}
/** Parameters of a function-like node (all such nodes carry `getParameters`). */
function getParams(fn: Node): ParameterDeclaration[] {
return (fn as unknown as { getParameters(): ParameterDeclaration[] }).getParameters();
}
function isAccessorReceiver(node: Node): boolean {
if (node.getKind() !== SyntaxKind.PropertyAccessExpression) return false;
return node.asKindOrThrow(SyntaxKind.PropertyAccessExpression).getName() === 'accessor';
}
/**
* Per-interface method declared return-type text. Lets the analyzer resolve
* what `agents.getHandle(...)` returns once it knows `agents: IAgentLifecycleService`.
*/
function collectInterfaceMethodReturns(
interfacesByName: Map<string, InterfaceDeclaration>,
): Map<string, Map<string, string>> {
const out = new Map<string, Map<string, string>>();
for (const [name, iface] of interfacesByName) {
const methods = new Map<string, string>();
for (const member of iface.getMembers()) {
if (member.getKind() === SyntaxKind.MethodSignature) {
const m = member.asKindOrThrow(SyntaxKind.MethodSignature);
const rt = m.getReturnTypeNode()?.getText();
if (rt) methods.set(m.getName(), rt);
}
}
out.set(name, methods);
}
return out;
}
/**
* Best-effort, parse-only type-text inference for an expression within a
* function. Handles just enough to follow scope-typed handles:
* - parameter / variable annotations,
* - `<x>.accessor.get(IToken)` the token (DI accessor returns its type),
* - `this.method(...)` the class method's declared return type,
* - `<base>.method(...)` the interface method's declared return type,
* - `await X` and single-step identifier aliases.
* Returns `undefined` when it can't tell callers fall back to source scope.
* `depth` bounds identifier-chasing so we don't loop on aliased locals.
*/
function inferExprTypeText(
expr: Node,
cls: ClassDeclaration,
ifaceMethods: Map<string, Map<string, string>>,
fn: Node,
depth = 0,
): string | undefined {
if (depth > 6) return undefined;
const kind = expr.getKind();
if (kind === SyntaxKind.AwaitExpression) {
const inner = (expr as unknown as { getExpression(): Node }).getExpression();
return inferExprTypeText(inner, cls, ifaceMethods, fn, depth + 1);
}
if (kind === SyntaxKind.AsExpression || kind === SyntaxKind.NonNullExpression) {
const inner = (expr as unknown as { getExpression(): Node }).getExpression();
return inferExprTypeText(inner, cls, ifaceMethods, fn, depth + 1);
}
if (kind === SyntaxKind.CallExpression) {
const call = expr.asKindOrThrow(SyntaxKind.CallExpression);
const callee = call.getExpression();
if (callee.getKind() !== SyntaxKind.PropertyAccessExpression) return undefined;
const pae = callee.asKindOrThrow(SyntaxKind.PropertyAccessExpression);
const methodName = pae.getName();
const base = pae.getExpression();
// `<x>.accessor.get(IToken)` → the DI accessor returns the token's type.
if (methodName === 'get' && isAccessorReceiver(base)) {
const first = call.getArguments()[0];
if (first && first.getKind() === SyntaxKind.Identifier) return first.getText();
return undefined;
}
// `this.method(...)` → class method's declared return type.
if (base.getKind() === SyntaxKind.ThisKeyword) {
return cls.getMethod(methodName)?.getReturnTypeNode()?.getText();
}
// `<base>.method(...)` → resolve base to an interface, look up the method.
const baseType = inferExprTypeText(base, cls, ifaceMethods, fn, depth + 1);
if (baseType === undefined) return undefined;
return ifaceMethods.get(stripTypeWrappers(baseType))?.get(methodName);
}
if (kind === SyntaxKind.Identifier) {
return resolveIdentifierTypeText(expr, cls, ifaceMethods, fn, depth + 1);
}
// `this.<field>` → the field's declared type (ctor parameter property or
// class property annotation).
if (kind === SyntaxKind.PropertyAccessExpression) {
const pae = expr.asKindOrThrow(SyntaxKind.PropertyAccessExpression);
if (pae.getExpression().getKind() === SyntaxKind.ThisKeyword) {
return thisFieldTypeText(cls, pae.getName());
}
return undefined;
}
// `a ?? b` → either branch's type.
if (kind === SyntaxKind.BinaryExpression) {
const bin = expr.asKindOrThrow(SyntaxKind.BinaryExpression);
if (bin.getOperatorToken().getKind() === SyntaxKind.QuestionQuestionToken) {
return (
inferExprTypeText(bin.getLeft(), cls, ifaceMethods, fn, depth + 1) ??
inferExprTypeText(bin.getRight(), cls, ifaceMethods, fn, depth + 1)
);
}
return undefined;
}
// `cond ? a : b` → either branch's type.
if (kind === SyntaxKind.ConditionalExpression) {
const cond = expr.asKindOrThrow(SyntaxKind.ConditionalExpression);
return (
inferExprTypeText(cond.getWhenTrue(), cls, ifaceMethods, fn, depth + 1) ??
inferExprTypeText(cond.getWhenFalse(), cls, ifaceMethods, fn, depth + 1)
);
}
return undefined;
}
function thisFieldTypeText(cls: ClassDeclaration, fieldName: string): string | undefined {
// Constructor parameter property: `constructor(@IX private readonly foo: IX)`.
const ctor = cls.getConstructors()[0];
if (ctor) {
for (const p of ctor.getParameters()) {
if (p.getName() !== fieldName) continue;
const t = p.getTypeNode()?.getText();
if (t) return t;
}
}
// Class property with an explicit type annotation.
return cls.getProperty(fieldName)?.getTypeNode()?.getText();
}
function resolveIdentifierTypeText(
id: Node,
cls: ClassDeclaration,
ifaceMethods: Map<string, Map<string, string>>,
fn: Node,
depth: number,
): string | undefined {
const name = id.getText();
for (const p of getParams(fn)) {
if (p.getName() === name) {
const t = p.getTypeNode()?.getText();
if (t) return t;
}
}
const decls = fn.getDescendantsOfKind(SyntaxKind.VariableDeclaration);
for (const decl of decls) {
if (decl.getName() !== name) continue;
if (decl.getStart() > id.getStart()) continue;
const annotated = decl.getTypeNode()?.getText();
if (annotated) return annotated;
const init = decl.getInitializer();
if (init) {
const inferred = inferExprTypeText(init, cls, ifaceMethods, fn, depth + 1);
if (inferred) return inferred;
}
}
return undefined;
}
/**
* For a `<obj>.accessor.get(IX)` call, return the scope of the handle `<obj>`
* when it can be inferred from a scope-typed handle alias. Returns `undefined`
* for the scope-agnostic cases (base `IScopeHandle`, unions, injected
* accessors) so the caller keeps the default source-scope resolution.
*/
function inferAccessorScope(
getCall: CallExpression,
cls: ClassDeclaration,
ifaceMethods: Map<string, Map<string, string>>,
): ServiceScope | undefined {
const getExpr = getCall.getExpression();
if (getExpr.getKind() !== SyntaxKind.PropertyAccessExpression) return undefined;
const receiver = getExpr.asKindOrThrow(SyntaxKind.PropertyAccessExpression).getExpression();
if (!isAccessorReceiver(receiver)) return undefined;
const obj = receiver.asKindOrThrow(SyntaxKind.PropertyAccessExpression).getExpression();
const fn = enclosingFunction(getCall);
if (fn === undefined) return undefined;
return handleScopeFromTypeText(inferExprTypeText(obj, cls, ifaceMethods, fn));
}
/**
* Pass 2 for a given impl class, walk method bodies and detect:
* - `<expr>.get(IToken)[.method(...)]` accessor edge (with optional `toMethod`)
@ -450,6 +736,7 @@ function collectRuntimeEdges(
source: ServiceNode,
injectedFields: Map<string, string>,
acc: EdgeAccumulator,
ifaceMethods: Map<string, Map<string, string>>,
): void {
const filePath = relFromRepo(cls.getSourceFile().getFilePath());
@ -476,7 +763,11 @@ function collectRuntimeEdges(
const toMethod = chainedMethodName(call);
const ref: EdgeRef = { ...baseRef };
if (toMethod !== undefined) ref.toMethod = toMethod;
pushEdge(acc, source.id, source, tokenName, 'accessor', ref);
// If this is `<handle>.accessor.get(IX)` and the handle's scope is
// statically known (e.g. `agent: IAgentScopeHandle`), resolve IX against
// that scope instead of the source service's scope.
const accessorScope = inferAccessorScope(call, cls, ifaceMethods);
pushEdge(acc, source.id, source, tokenName, 'accessor', ref, accessorScope);
continue;
}
@ -542,6 +833,7 @@ export function analyze(options: { srcRoot?: string; generatedAt?: string } = {}
const { services, implClasses, bindings } = collectServices(sourceFiles);
const interfacesByName = collectInterfaces(sourceFiles);
const ifaceMethods = collectInterfaceMethodReturns(interfacesByName);
// Seed the framework tokens as synthetic nodes so edges to them resolve
// like any other registered service. They are marked domain=`framework`
@ -630,7 +922,7 @@ export function analyze(options: { srcRoot?: string; generatedAt?: string } = {}
if (dep.token === svc.token) continue;
pushEdge(acc, svc.id, svc, dep.token, 'ctor', { file: filePath, line: dep.line });
}
collectRuntimeEdges(cls, svc, injectedFields, acc);
collectRuntimeEdges(cls, svc, injectedFields, acc, ifaceMethods);
}
// Synthesise interface-only nodes for tokens referenced by edges but with no
@ -680,6 +972,35 @@ export function analyze(options: { srcRoot?: string; generatedAt?: string } = {}
services.push(node);
}
// Synthesise scope-mismatch nodes: tokens that ARE registered but referenced
// from a scope that can't see them. Placed at the token's real registered
// scope (from `actualScope`) and flagged so the viewer styles them apart
// from genuinely missing implementations.
const mismatchTokens = new Map<string, ServiceScope>();
for (const edge of acc.edges.values()) {
if (!edge.scopeMismatch || edge.actualScope === undefined) continue;
if (!mismatchTokens.has(edge.token)) mismatchTokens.set(edge.token, edge.actualScope);
}
for (const [token, scope] of mismatchTokens) {
const registered = acc.bindings.get(token)?.get(scope);
const node: ServiceNode = {
id: `scopeMismatch::${token}`,
token,
impl: token,
scope,
domain: registered?.domain ?? 'unknown',
file: '',
line: 0,
scopeMismatch: true,
};
const iface = interfacesByName.get(token);
if (iface) {
const members = collectInterfaceMembers(iface);
if (members.length > 0) node.publicMembers = members;
}
services.push(node);
}
return {
generatedAt: options.generatedAt ?? new Date(0).toISOString(),
services: services.sort(

View file

@ -15,9 +15,9 @@ export type EdgeKind =
| 'publish'
/** `<eventBus>.subscribe(...)` — subscribes to `IEventService`. */
| 'subscribe'
/** `<eventSink>.emit(...)` — emits on `IAgentEventSinkService`. */
/** `<record>.signal(...)` / `<record>.append(...)` — emits on `IAgentRecordService`. */
| 'emit'
/** `<eventSink>.on(...)` — listens on `IAgentEventSinkService`. */
/** `<record>.on(...)` — listens on `IAgentRecordService`. */
| 'on';
export interface ServiceNode {
@ -57,6 +57,13 @@ export interface ServiceNode {
* services rather than being dropped as dangling edges.
*/
unresolved?: true;
/**
* True for synthesized scope-mismatch nodes: the token IS registered, but at
* a scope invisible to the edge's source. Rendered distinctly (and placed at
* the token's real registered scope) so a cross-scope reach reads differently
* from a genuinely missing implementation.
*/
scopeMismatch?: true;
}
export interface EdgeRef {
@ -90,18 +97,29 @@ export interface Edge {
* Resolved target `ServiceNode.id` the concrete registration that the
* DI container would actually pick when the source is instantiated. For
* `unresolved: true` edges this is the token that couldn't be resolved,
* prefixed with `unresolved::` for visualisation purposes.
* prefixed with `unresolved::`; for `scopeMismatch: true` edges it is
* prefixed with `scopeMismatch::`.
*/
to: string;
/** The interface/decorator name that appears at the source site. */
token: string;
kind: EdgeKind;
/**
* True when there is no impl registered for `token` at a scope visible
* from the source (walking source scope up to root). A `ctor` edge in
* this state would crash the container at instantiation time.
* True when there is no impl registered for `token` at ANY scope the
* token is simply unknown to the container. A `ctor` edge in this state
* would crash the container at instantiation time.
*/
unresolved?: true;
/**
* True when the token IS registered, but only at a scope that is not
* visible from the source (e.g. an App-scope service reaching for an
* Agent-scope token through an accessor whose scope the analyzer couldn't
* pin down). Distinct from `unresolved`: an implementation exists, the
* edge just can't be satisfied from where it is requested.
*/
scopeMismatch?: true;
/** When `scopeMismatch`, the innermost scope where `token` is registered. */
actualScope?: ServiceScope;
/** One or more locations that produced this edge (deduped). */
refs: EdgeRef[];
}

View file

@ -21,14 +21,14 @@ import {
import { linkAbortSignal, userCancellationReason } from '#/_base/utils/abort';
import { IAgentLifecycleService } from '#/session/agent-lifecycle';
import type { IScopeHandle } from '#/_base/di/scope';
import type { IAgentScopeHandle } from '#/_base/di/scope';
import {
IAgentContextMemoryService,
type ContextMessage,
type PromptOrigin,
} from '#/agent/contextMemory';
import { ErrorCodes, toKimiErrorPayload, type KimiErrorPayload } from '#/errors';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import { IAgentExternalHooksService } from '#/agent/externalHooks';
import { isAbortError } from '#/agent/loop/errors';
import { IAgentProfileService } from '#/agent/profile';
@ -183,7 +183,7 @@ export function cancelAllChildren(
async function ensureParent(
lifecycle: IAgentLifecycleService,
parentAgentId: string,
): Promise<IScopeHandle> {
): Promise<IAgentScopeHandle> {
const existing = lifecycle.getHandle(parentAgentId);
if (existing !== undefined) return existing;
throw new Error(`Parent agent "${parentAgentId}" does not exist`);
@ -194,7 +194,7 @@ async function requireChild(
parentAgentId: string,
metadata: ISessionMetadata | undefined,
agentId: string,
): Promise<IScopeHandle> {
): Promise<IAgentScopeHandle> {
if (metadata !== undefined) {
const meta = (await metadata.read()).agents?.[agentId];
if (meta === undefined) throw new Error(`Agent instance "${agentId}" does not exist`);
@ -212,13 +212,13 @@ async function requireChild(
return child;
}
function ensureAgentTool(child: IScopeHandle): void {
function ensureAgentTool(child: IAgentScopeHandle): void {
// Force-instantiate the child agent's `Agent` tool registrar so its `Agent`
// tool is registered before the child's first turn builds its tool list.
child.accessor.get(IAgentToolService);
}
function configureChild(parent: IScopeHandle, child: IScopeHandle, profileName: string): void {
function configureChild(parent: IAgentScopeHandle, child: IAgentScopeHandle, profileName: string): void {
const parentProfile = parent.accessor.get(IAgentProfileService);
const childProfile = child.accessor.get(IAgentProfileService);
const parentData = parentProfile.data();
@ -241,13 +241,13 @@ function configureChild(parent: IScopeHandle, child: IScopeHandle, profileName:
}
function emitSpawned(
parent: IScopeHandle,
parent: IAgentScopeHandle,
parentAgentId: string,
subagentId: string,
profileName: string,
options: RunSubagentOptions,
): void {
parent.accessor.get(IAgentEventSinkService)?.emit({
parent.accessor.get(IAgentRecordService)?.signal({
type: 'subagent.spawned',
subagentId,
subagentName: profileName,
@ -264,17 +264,17 @@ function emitSpawned(
});
}
function emitStarted(parent: IScopeHandle, subagentId: string): void {
parent.accessor.get(IAgentEventSinkService)?.emit({ type: 'subagent.started', subagentId });
function emitStarted(parent: IAgentScopeHandle, subagentId: string): void {
parent.accessor.get(IAgentRecordService)?.signal({ type: 'subagent.started', subagentId });
}
function emitCompleted(
parent: IScopeHandle,
parent: IAgentScopeHandle,
subagentId: string,
resultSummary: string,
usage?: TokenUsage,
): void {
parent.accessor.get(IAgentEventSinkService)?.emit({
parent.accessor.get(IAgentRecordService)?.signal({
type: 'subagent.completed',
subagentId,
resultSummary,
@ -283,14 +283,14 @@ function emitCompleted(
}
function emitFailed(
parent: IScopeHandle,
parent: IAgentScopeHandle,
subagentId: string,
error: unknown,
options: RunSubagentOptions,
): void {
if (isAbortError(error)) return;
if (shouldSuppressQueuedAttemptFailureEvent(options, error)) return;
parent.accessor.get(IAgentEventSinkService)?.emit({
parent.accessor.get(IAgentRecordService)?.signal({
type: 'subagent.failed',
subagentId,
error: errorMessage(error),
@ -299,7 +299,7 @@ function emitFailed(
async function triggerSubagentStart(
parent: IScopeHandle,
parent: IAgentScopeHandle,
profileName: string,
prompt: string,
signal: AbortSignal,
@ -313,7 +313,7 @@ async function triggerSubagentStart(
);
}
function triggerSubagentStop(parent: IScopeHandle, profileName: string, result: string): void {
function triggerSubagentStop(parent: IAgentScopeHandle, profileName: string, result: string): void {
parent.accessor.get(IAgentExternalHooksService)?.triggerSubagentStop({
agentName: profileName,
response: result.slice(0, HOOK_TEXT_PREVIEW_LENGTH),
@ -327,9 +327,9 @@ function observeFirstRequest(turn: Turn, options: RunSubagentOptions): void {
async function runWithActiveChild(
parentAgentId: string,
child: IScopeHandle,
child: IAgentScopeHandle,
options: RunSubagentOptions,
parent: IScopeHandle,
parent: IAgentScopeHandle,
profileName: string,
run: (
turn: { current?: Turn },
@ -359,8 +359,8 @@ async function runWithActiveChild(
}
async function runPromptTurn(
child: IScopeHandle,
parent: IScopeHandle,
child: IAgentScopeHandle,
parent: IAgentScopeHandle,
options: RunSubagentOptions,
profileName: string,
turnRef: { current?: Turn },
@ -389,8 +389,8 @@ async function runPromptTurn(
}
async function runRetryTurn(
child: IScopeHandle,
parent: IScopeHandle,
child: IAgentScopeHandle,
parent: IAgentScopeHandle,
options: RunSubagentOptions,
profileName: string,
turnRef: { current?: Turn },
@ -429,7 +429,7 @@ async function awaitTurn(
}
async function completeSummary(
child: IScopeHandle,
child: IAgentScopeHandle,
controller: AbortController,
turnRef: { current?: Turn },
): Promise<string> {

View file

@ -4,9 +4,9 @@
* Owns the agent's registry of running and restored background tasks:
* registers and drives tasks to completion, retains a bounded output ring,
* persists task state and output through `background` persistence, reads
* limits through `config`, records lifecycle through `wireRecord`, delivers
* terminal notifications through `contextMemory`, and broadcasts through
* `eventSink`. Bound at Agent scope.
* limits through `config`, records lifecycle and broadcasts through `record`,
* and delivers terminal notifications through `contextMemory`. Bound at Agent
* scope.
*/
import { randomBytes } from 'node:crypto';
@ -27,15 +27,13 @@ import {
import { IAgentContextMemoryService } from '#/agent/contextMemory';
import { IConfigService } from '#/app/config';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentExternalHooksService } from '#/agent/externalHooks';
import { IAgentPromptService } from '#/agent/prompt';
import { ISessionContext } from '#/session/session-context';
import { IAtomicDocumentStore, IStorageService } from '#/app/storage';
import { ITelemetryService } from '#/app/telemetry';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import type { WireRecord } from '#/agent/wireRecord';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentRecordService, type AgentRecord } from '#/agent/record';
import {
IAgentBackgroundService,
type BackgroundLoadOptions,
@ -133,8 +131,7 @@ export class AgentBackgroundService extends Disposable implements IAgentBackgrou
private readonly persistence: BackgroundTaskPersistence;
constructor(
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentRecordService private readonly record: IAgentRecordService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@IAgentPromptService private readonly prompt: IAgentPromptService,
@IAgentExternalHooksService private readonly externalHooks: IAgentExternalHooksService,
@ -153,17 +150,21 @@ export class AgentBackgroundService extends Disposable implements IAgentBackgrou
byteStore,
);
this._register(
wireRecord.register('background.task.started', (record) => {
this.applyRestoredTask(record);
record.define('background.task.started', {
resume: (r) => {
this.applyRestoredTask(r);
},
}),
);
this._register(
wireRecord.register('background.task.terminated', (record) => {
this.applyRestoredTask(record);
record.define('background.task.terminated', {
resume: (r) => {
this.applyRestoredTask(r);
},
}),
);
this._register(
wireRecord.hooks.onResumeEnded.register(
record.hooks.onResumeEnded.register(
'background-lifecycle-resume',
async (_ctx, next) => {
await this.loadFromDisk({ replace: false });
@ -561,7 +562,7 @@ export class AgentBackgroundService extends Disposable implements IAgentBackgrou
}
private applyRestoredTask(
record: WireRecord<'background.task.started' | 'background.task.terminated'>,
record: AgentRecord<'background.task.started' | 'background.task.terminated'>,
): void {
const info = record.info;
if (this.tasks.has(info.taskId)) return;
@ -685,16 +686,16 @@ export class AgentBackgroundService extends Disposable implements IAgentBackgrou
}
private recordTaskStarted(info: BackgroundTaskInfo): void {
this.wireRecord.append({ type: 'background.task.started', info });
this.events.emit({ type: 'background.task.started', info });
this.record.append({ type: 'background.task.started', info });
this.record.signal({ type: 'background.task.started', info });
this.telemetry.track('background_task_created', {
kind: info.kind === 'process' ? 'bash' : info.kind,
});
}
private recordTaskTerminated(info: BackgroundTaskInfo): void {
this.wireRecord.append({ type: 'background.task.terminated', info });
this.events.emit({ type: 'background.task.terminated', info });
this.record.append({ type: 'background.task.terminated', info });
this.record.signal({ type: 'background.task.terminated', info });
this.telemetry.track('background_task_completed', {
kind: info.kind,
duration: info.endedAt !== null ? info.endedAt - info.startedAt : null,

View file

@ -3,7 +3,7 @@ import {
} from "#/_base/di";
import { OrderedHookSlot } from '#/hooks';
import { IAgentReplayBuilderService } from '#/agent/replayBuilder';
import { IAgentWireRecordService, type WireRecord } from '#/agent/wireRecord';
import { IAgentRecordService, type AgentRecord } from '#/agent/record';
import { IAgentContextMemoryService } from './contextMemory';
import { ensureMessageId } from './messageId';
import type { ContextMessage } from './types';
@ -35,18 +35,17 @@ export class AgentContextMemoryService extends Disposable implements IAgentConte
};
constructor(
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService,
) {
super();
this._register(
wireRecord.register(
'context.splice',
(record) => {
this.applySplice(record);
record.define('context.splice', {
resume: (r) => {
this.applySplice(r);
},
{
blobs: (record) => record.messages.map((message, index) => ({
blobs: (r) =>
r.messages.map((message, index) => ({
parts: message.content,
replace: (current, content) => ({
...current,
@ -55,8 +54,7 @@ export class AgentContextMemoryService extends Disposable implements IAgentConte
),
}),
})),
},
),
}),
);
}
@ -71,18 +69,18 @@ export class AgentContextMemoryService extends Disposable implements IAgentConte
tokens?: number,
): void {
const stamped = messages.map(ensureMessageId);
const record: WireRecord<'context.splice'> = {
const record: AgentRecord<'context.splice'> = {
type: 'context.splice',
start,
deleteCount,
messages: stamped,
tokens,
};
this.wireRecord.append(record);
this.record.append(record);
this.applySplice(record);
}
private applySplice(record: WireRecord<'context.splice'>): void {
private applySplice(record: AgentRecord<'context.splice'>): void {
const removedMessages =
record.deleteCount > 0 && record.start > 0
? this.history.slice(record.start, record.start + record.deleteCount)

View file

@ -6,8 +6,7 @@ import {
} from "#/_base/utils/tokens";
import type { ContextMessage } from '#/agent/contextMemory';
import { IAgentContextMemoryService } from '#/agent/contextMemory';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentWireRecordService, type WireRecord } from '#/agent/wireRecord';
import { IAgentRecordService, type AgentRecord } from '#/agent/record';
import {
IAgentContextSizeService,
type ContextSizeStatus,
@ -39,8 +38,7 @@ export class AgentContextSizeService
constructor(
@IAgentContextMemoryService private readonly context: IAgentContextMemoryService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentRecordService private readonly records: IAgentRecordService,
) {
super();
this._register(
@ -50,8 +48,10 @@ export class AgentContextSizeService
}),
);
this._register(
wireRecord.register('context_size.measured', (record) => {
this.applyMeasurement(record);
records.define('context_size.measured', {
resume: (r) => {
this.applyMeasurement(r);
},
}),
);
}
@ -66,12 +66,12 @@ export class AgentContextSizeService
}
measured(length: number, tokens: number): void {
const record: WireRecord<'context_size.measured'> = {
const record: AgentRecord<'context_size.measured'> = {
type: 'context_size.measured',
length,
tokens,
};
this.wireRecord.append(record);
this.records.append(record);
this.applyMeasurement(record);
}
@ -102,7 +102,7 @@ export class AgentContextSizeService
this.emitIfChanged();
}
private applyMeasurement(record: WireRecord<'context_size.measured'>): void {
private applyMeasurement(record: AgentRecord<'context_size.measured'>): void {
const length = clampMeasuredLength(record.length, this.estimates.length);
const tokens = Math.max(0, record.tokens);
this.measuredPrefixTokens[length] = tokens;
@ -125,7 +125,7 @@ export class AgentContextSizeService
return;
}
this.lastEmitted = status;
this.events.emit({
this.records.signal({
type: 'agent.status.updated',
contextTokens: status.contextTokens,
});

View file

@ -22,12 +22,11 @@ import { IConfigService } from '#/app/config';
import { IAtomicDocumentStore } from '#/app/storage';
import { ITelemetryService } from '#/app/telemetry';
import type { ContextMessage } from '#/agent/contextMemory';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentPromptService } from '#/agent/prompt';
import { IAgentRecordService } from '#/agent/record';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import type { Turn } from '#/agent/turn';
import { IAgentTurnService } from '#/agent/turn';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import {
type CronConfig,
@ -145,8 +144,7 @@ export class AgentCronService extends Disposable implements IAgentCronService {
constructor(
options: CronOptions = {},
@IAgentPromptService private readonly prompt: IAgentPromptService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentTurnService private readonly turnService: IAgentTurnService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@IAgentToolRegistryService private readonly toolRegistry: IAgentToolRegistryService,
@ -167,22 +165,28 @@ export class AgentCronService extends Disposable implements IAgentCronService {
resolveClockSources(this.cronConfig.clock, this.cronConfig.debug) ?? SYSTEM_CLOCKS;
this._register(
wireRecord.register('cron.add', (record) => {
if (this.enabled) this.adopt(record.task);
record.define('cron.add', {
resume: (r) => {
if (this.enabled) this.adopt(r.task);
},
}),
);
this._register(
wireRecord.register('cron.delete', (record) => {
if (this.enabled) this.removeByIds(record.ids);
record.define('cron.delete', {
resume: (r) => {
if (this.enabled) this.removeByIds(r.ids);
},
}),
);
this._register(
wireRecord.register('cron.cursor', (record) => {
if (this.enabled) this.markFired(record.id, record.lastFiredAt);
record.define('cron.cursor', {
resume: (r) => {
if (this.enabled) this.markFired(r.id, r.lastFiredAt);
},
}),
);
this._register(
wireRecord.hooks.onResumeEnded.register('cron-lifecycle-resume', async (_ctx, next) => {
record.hooks.onResumeEnded.register('cron-lifecycle-resume', async (_ctx, next) => {
await this.loadFromDisk({ replace: false });
this.start();
await next();
@ -221,7 +225,7 @@ export class AgentCronService extends Disposable implements IAgentCronService {
createdAt: this.clocks.wallNow(),
};
this.tasks.set(task.id, task);
this.wireRecord.append({ type: 'cron.add', task });
this.record.append({ type: 'cron.add', task });
this.persistEnqueue(task.id, () => this.atomicDocs.set(CRON_SCOPE, cronKey(task.id), task));
return task;
}
@ -230,7 +234,7 @@ export class AgentCronService extends Disposable implements IAgentCronService {
const removed = this.removeByIds(ids);
if (removed.length === 0) return removed;
this.wireRecord.append({ type: 'cron.delete', ids: removed });
this.record.append({ type: 'cron.delete', ids: removed });
for (const id of removed) {
this.persistEnqueue(id, () => this.atomicDocs.delete(CRON_SCOPE, cronKey(id)));
}
@ -457,7 +461,7 @@ export class AgentCronService extends Disposable implements IAgentCronService {
toolCalls: [],
origin,
};
this.events.emit({ type: 'cron.fired', origin, prompt: task.prompt });
this.record.signal({ type: 'cron.fired', origin, prompt: task.prompt });
const turn = this.prompt.steer(message);
this.telemetry.track(CRON_FIRED, {
recurring: task.recurring !== false,
@ -472,7 +476,7 @@ export class AgentCronService extends Disposable implements IAgentCronService {
const updated = this.markFired(id, lastFiredAt);
if (updated === undefined) return;
this.wireRecord.append({ type: 'cron.cursor', id, lastFiredAt });
this.record.append({ type: 'cron.cursor', id, lastFiredAt });
this.persistEnqueue(id, () => this.atomicDocs.set(CRON_SCOPE, cronKey(id), updated));
}

View file

@ -1,11 +0,0 @@
import type { IDisposable } from "#/_base/di";
import { createDecorator } from "#/_base/di";
import type { AgentEvent } from '@moonshot-ai/protocol';
export interface IAgentEventSinkService {
readonly _serviceBrand: undefined;
emit(event: AgentEvent): void;
on(handler: (event: AgentEvent) => void): IDisposable;
}
export const IAgentEventSinkService = createDecorator<IAgentEventSinkService>('agentEventSinkService');

View file

@ -1,37 +0,0 @@
import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import {
Disposable,
} from "#/_base/di";
import { Emitter } from "#/_base/event";
import { IAgentWireRecordService } from '#/agent/wireRecord';
import type { AgentEvent } from '@moonshot-ai/protocol';
import { IAgentEventSinkService } from './eventSink';
export class AgentEventSinkService extends Disposable implements IAgentEventSinkService {
declare readonly _serviceBrand: undefined;
private readonly onDidEmitEmitter = this._register(new Emitter<AgentEvent>());
constructor(@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService) {
super();
}
emit(event: AgentEvent): void {
if (this.wireRecord.restoring) return;
this.onDidEmitEmitter.fire(event);
}
on(handler: (event: AgentEvent) => void) {
return this.onDidEmitEmitter.event(handler);
}
}
registerScopedService(
LifecycleScope.Agent,
IAgentEventSinkService,
AgentEventSinkService,
InstantiationType.Delayed,
'eventSink',
);

View file

@ -1,6 +0,0 @@
/**
* `eventBus` domain barrel - re-exports the eventBus service contract and implementation.
*/
export * from './eventSink';
export * from './eventSinkService';

View file

@ -18,7 +18,6 @@ import { estimateTokens, estimateTokensForMessages } from "#/_base/utils/tokens"
import { IAgentContextMemoryService } from '#/agent/contextMemory';
import { IAgentContextProjectorService } from '#/agent/contextProjector';
import { IAgentContextSizeService } from '#/agent/contextSize';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentExternalHooksService } from '#/agent/externalHooks';
import { IAgentLLMRequesterService, type LLMEvent } from '#/agent/llmRequester';
import { isAbortError } from '#/agent/loop/errors';
@ -29,7 +28,7 @@ import { ITelemetryService } from '#/app/telemetry';
import { IAgentToolStoreService } from '#/agent/toolStore';
import { IAgentTurnService, type TurnContextOverflowContext } from '#/agent/turn';
import type { ContextMessage } from '#/agent/contextMemory';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentRecordService } from '#/agent/record';
import {
TODO_STORE_KEY,
renderTodoList,
@ -104,8 +103,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
@IAgentProfileService private readonly profile: IAgentProfileService,
@IAgentToolStoreService private readonly toolStore: IAgentToolStoreService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService,
@IAgentExternalHooksService private readonly externalHooks: IAgentExternalHooksService,
@IAgentTurnService turnService: IAgentTurnService,
@ -138,32 +136,38 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
}),
);
this._register(
wireRecord.register('full_compaction.begin', (record) => {
this.replayBuilder.push({
type: 'compaction',
instruction: record.instruction,
});
record.define('full_compaction.begin', {
resume: (r) => {
this.replayBuilder.push({
type: 'compaction',
instruction: r.instruction,
});
},
}),
);
this._register(
wireRecord.register('full_compaction.cancel', () => {
this.replayBuilder.patchLast('compaction', { result: 'cancelled' });
record.define('full_compaction.cancel', {
resume: () => {
this.replayBuilder.patchLast('compaction', { result: 'cancelled' });
},
}),
);
this._register(
wireRecord.register('full_compaction.complete', (record) => {
const message = compactionSummaryMessage(this.context.get());
if (message === undefined) return;
const summary = contextMessageText(message);
this.replayBuilder.removeLastMessages(new Set([message]));
this.replayBuilder.patchLast('compaction', {
result: {
summary,
compactedCount: record.compactedCount,
tokensBefore: record.tokensBefore,
tokensAfter: record.tokensAfter,
},
});
record.define('full_compaction.complete', {
resume: (r) => {
const message = compactionSummaryMessage(this.context.get());
if (message === undefined) return;
const summary = contextMessageText(message);
this.replayBuilder.removeLastMessages(new Set([message]));
this.replayBuilder.patchLast('compaction', {
result: {
summary,
compactedCount: r.compactedCount,
tokensBefore: r.tokensBefore,
tokensAfter: r.tokensAfter,
},
});
},
}),
);
}
@ -188,8 +192,8 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
throw new KimiError(ErrorCodes.COMPACTION_UNABLE, 'No prefix that can be compacted in current history.');
}
this.wireRecord.append({ type: 'full_compaction.begin', ...data });
this.events.emit({
this.record.append({ type: 'full_compaction.begin', ...data });
this.record.signal({
type: 'compaction.started',
trigger: data.source,
instruction: data.instruction,
@ -208,15 +212,15 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
cancel(): void {
const active = this.compacting;
if (active === null) return;
this.wireRecord.append({ type: 'full_compaction.cancel' });
this.record.append({ type: 'full_compaction.cancel' });
active.abortController.abort();
this.compacting = null;
this.events.emit({ type: 'compaction.cancelled' });
this.record.signal({ type: 'compaction.cancelled' });
}
private markCompleted(result: FullCompactionCompleteData): void {
if (this.compacting === null) return;
this.wireRecord.append({ type: 'full_compaction.complete', ...result });
this.record.append({ type: 'full_compaction.complete', ...result });
this.compacting = null;
}
@ -281,7 +285,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
}
}, { once: true });
}
this.events.emit({ type: 'compaction.blocked', turnId });
this.record.signal({ type: 'compaction.blocked', turnId });
await active.promise;
}
@ -318,7 +322,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
if (this.compacting !== active) return;
this.markCompleted(completeData(finalResult));
this.events.emit({ type: 'compaction.completed', result: finalResult });
this.record.signal({ type: 'compaction.completed', result: finalResult });
this.externalHooks.triggerPostCompact({
trigger: data.source,
estimatedTokenCount: finalResult.tokensAfter,
@ -332,7 +336,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
if (blockedByTurn) {
throw error;
}
this.events.emit({
this.record.signal({
type: 'error',
...toKimiErrorPayload(error),
});

View file

@ -8,15 +8,13 @@ import {
} from "#/_base/di";
import { ErrorCodes, KimiError } from "#/errors";
import { IAgentContextInjectorService } from '#/agent/contextInjector';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentPermissionModeService } from '#/agent/permissionMode';
import { IAgentReplayBuilderService } from '#/agent/replayBuilder';
import { IAgentSystemReminderService } from '#/agent/systemReminder';
import type { TelemetryProperties } from '#/app/telemetry';
import { ITelemetryService } from '#/app/telemetry';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import type { WireRecord } from '#/agent/wireRecord';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentRecordService, type AgentRecord } from '#/agent/record';
import {
IAgentGoalService,
type GoalReasonInput,
@ -101,8 +99,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
constructor(
private readonly options: GoalServiceOptions = {},
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentSystemReminderService private readonly reminders: IAgentSystemReminderService,
@IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@ -121,27 +118,35 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
),
);
this._register(
wireRecord.register('forked', (record) => {
this.restoreForked(record);
record.define('forked', {
resume: (r) => {
this.restoreForked(r);
},
}),
);
this._register(
wireRecord.register('goal.create', (record) => {
this.restoreCreate(record);
record.define('goal.create', {
resume: (r) => {
this.restoreCreate(r);
},
}),
);
this._register(
wireRecord.register('goal.update', (record) => {
this.restoreUpdate(record);
record.define('goal.update', {
resume: (r) => {
this.restoreUpdate(r);
},
}),
);
this._register(
wireRecord.register('goal.clear', () => {
this.restoreClear();
record.define('goal.clear', {
resume: () => {
this.restoreClear();
},
}),
);
this._register(
wireRecord.hooks.onResumeEnded.register('goal-normalize-after-replay', async (_ctx, next) => {
record.hooks.onResumeEnded.register('goal-normalize-after-replay', async (_ctx, next) => {
await next();
this.normalizeAfterReplay();
}),
@ -207,7 +212,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
};
this.persistState(state);
this.wireRecord.append({
this.record.append({
type: 'goal.create',
goalId: state.goalId,
objective: state.objective,
@ -378,7 +383,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
this.appendStatusUpdate(state, 'runtime', reason);
}
private restoreCreate(record: WireRecord<'goal.create'>): void {
private restoreCreate(record: AgentRecord<'goal.create'>): void {
const state: GoalState = {
goalId: record.goalId,
objective: record.objective,
@ -397,7 +402,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
});
}
private restoreUpdate(record: WireRecord<'goal.update'>): void {
private restoreUpdate(record: AgentRecord<'goal.update'>): void {
const state = this.state;
if (state === undefined) return;
@ -441,7 +446,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
this.state = undefined;
}
private restoreForked(_record: WireRecord<'forked'>): void {
private restoreForked(_record: AgentRecord<'forked'>): void {
const hadGoal = this.state !== undefined;
this.state = undefined;
if (!hadGoal) return;
@ -457,7 +462,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
): void {
if (this.state === undefined) return;
this.persistState(undefined, { silent: opts.emit === false });
this.wireRecord.append({ type: 'goal.clear' });
this.record.append({ type: 'goal.clear' });
if (opts.track !== false) this.telemetry.track('goal_cleared', { actor });
}
@ -479,9 +484,9 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
}
private appendGoalUpdate(
update: Omit<WireRecord<'goal.update'>, 'type' | 'time'>,
update: Omit<AgentRecord<'goal.update'>, 'type' | 'time'>,
): void {
this.wireRecord.append({
this.record.append({
type: 'goal.update',
...update,
});
@ -518,7 +523,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
}
private emitGoalUpdated(snapshot: GoalSnapshot | null, change?: GoalChange): void {
this.events.emit({ type: 'goal.updated', snapshot, change });
this.record.signal({ type: 'goal.updated', snapshot, change });
}
private statsOf(state: GoalState): GoalChangeStats {

View file

@ -136,66 +136,8 @@ export type LoopLiveOnlyEvent =
| LoopToolProgressEvent;
export type LoopEvent = LoopRecordedEvent | LoopLiveOnlyEvent;
export type LoopLiveEventEmitter = (event: LoopEvent) => void;
export type LoopEventDispatcher = {
(event: LoopRecordedEvent): Promise<void>;
(event: LoopLiveOnlyEvent): void;
};
export interface CreateLoopEventDispatcherInput {
readonly appendTranscriptRecord: (record: LoopRecordedEvent) => Promise<void>;
readonly emitLiveEvent?: LoopLiveEventEmitter | undefined;
}
export function createLoopEventDispatcher(
input: CreateLoopEventDispatcherInput,
): LoopEventDispatcher {
function dispatchEvent(event: LoopRecordedEvent): Promise<void>;
function dispatchEvent(event: LoopLiveOnlyEvent): void;
function dispatchEvent(event: LoopEvent): Promise<void> | void {
if (isRecordedEvent(event)) {
return recordEvent(input, event);
}
safeEmitLive(input.emitLiveEvent, event);
}
return dispatchEvent;
}
function isRecordedEvent(event: LoopEvent): event is LoopRecordedEvent {
return (
event.type === 'step.begin' ||
event.type === 'step.end' ||
event.type === 'content.part' ||
event.type === 'tool.call' ||
event.type === 'tool.result'
);
}
async function recordEvent(
input: CreateLoopEventDispatcherInput,
event: LoopRecordedEvent,
): Promise<void> {
await input.appendTranscriptRecord(event);
safeEmitLive(input.emitLiveEvent, event);
}
function safeEmitLive(emit: LoopLiveEventEmitter | undefined, event: LoopEvent): void {
if (emit === undefined) return;
let maybePromise: unknown;
try {
maybePromise = (emit as (event: LoopEvent) => unknown)(event);
} catch {
return;
}
if (
maybePromise !== undefined &&
maybePromise !== null &&
typeof (maybePromise as { then?: unknown }).then === 'function' &&
typeof (maybePromise as { catch?: unknown }).catch === 'function'
) {
(maybePromise as Promise<unknown>).catch(() => {
// Live listeners are best-effort; their failures must not affect the turn.
});
}
}

View file

@ -27,7 +27,6 @@ export type {
} from './types';
export type {
CreateLoopEventDispatcherInput,
LoopContentPartEvent,
LoopRecordedEvent,
LoopStepBeginEvent,
@ -36,7 +35,6 @@ export type {
LoopLiveOnlyEvent,
LoopEvent,
LoopInterruptReason,
LoopLiveEventEmitter,
LoopEventDispatcher,
LoopTextDeltaEvent,
LoopThinkingDeltaEvent,
@ -46,7 +44,6 @@ export type {
LoopToolResultEvent,
LoopTurnInterruptedEvent,
} from './events';
export { createLoopEventDispatcher } from './events';
export type {
LLM,

View file

@ -40,7 +40,7 @@ import { canonicalTelemetryArgs } from '#/_base/utils/canonical-args';
import { IAgentContextMemoryService, newMessageId, type ContextMessage } from '#/agent/contextMemory';
import { IAgentContextProjectorService } from '#/agent/contextProjector';
import { IAgentContextSizeService } from '#/agent/contextSize';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import { IAgentExternalHooksService } from '#/agent/externalHooks';
import { IAgentLLMRequesterService } from '#/agent/llmRequester';
import { ILogService } from '#/app/log';
@ -50,7 +50,6 @@ import { ITelemetryService } from '#/app/telemetry';
import { IAgentToolExecutorService } from '#/agent/toolExecutor';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import type { Turn, TurnResult } from '#/agent/turn';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import type {
LoopEvent,
LoopEventDispatcher,
@ -93,12 +92,11 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
@IAgentContextProjectorService private readonly projector: IAgentContextProjectorService,
@IAgentContextSizeService private readonly contextSize: IAgentContextSizeService,
@IAgentLLMRequesterService private readonly llmRequester: IAgentLLMRequesterService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentToolRegistryService private readonly toolRegistry: IAgentToolRegistryService,
@IAgentToolExecutorService private readonly toolExecutor: IAgentToolExecutorService,
@IAgentProfileService private readonly profile: IAgentProfileService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentExternalHooksService private readonly externalHooks: IAgentExternalHooksService,
@IConfigService private readonly config: IConfigService,
@ILogService private readonly log: ILogService,
@ -110,7 +108,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
}
await next();
});
this.wireRecord.hooks.onResumeEnded.register(
this.record.hooks.onResumeEnded.register(
'loop-service-finish-resume',
async (_event, next) => {
this.finishResume();
@ -248,7 +246,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
switch (event.type) {
case 'step.begin':
this.beginTrackedStep(event.step);
this.events.emit({
this.record.signal({
type: 'turn.step.started',
turnId: Number(event.turnId),
step: event.step,
@ -256,7 +254,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
});
return;
case 'step.end':
this.events.emit({
this.record.signal({
type: 'turn.step.completed',
turnId: Number(event.turnId),
step: event.step,
@ -274,7 +272,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
});
return;
case 'step.retrying':
this.events.emit({
this.record.signal({
type: 'turn.step.retrying',
turnId: Number(event.turnId),
step: event.step,
@ -295,7 +293,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
}
}
if (this.protocolTurnId === undefined || event.activeStep === undefined) return;
this.events.emit({
this.record.signal({
type: 'turn.step.interrupted',
turnId: this.protocolTurnId,
step: event.activeStep,
@ -305,7 +303,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
return;
case 'text.delta':
if (this.protocolTurnId === undefined) return;
this.events.emit({
this.record.signal({
type: 'assistant.delta',
turnId: this.protocolTurnId,
delta: event.delta,
@ -313,7 +311,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
return;
case 'thinking.delta':
if (this.protocolTurnId === undefined) return;
this.events.emit({
this.record.signal({
type: 'thinking.delta',
turnId: this.protocolTurnId,
delta: event.delta,
@ -321,7 +319,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
return;
case 'tool.call.delta':
if (this.protocolTurnId === undefined) return;
this.events.emit({
this.record.signal({
type: 'tool.call.delta',
turnId: this.protocolTurnId,
toolCallId: event.toolCallId,
@ -331,7 +329,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
return;
case 'tool.call':
this.trackToolCallStarted(event);
this.events.emit({
this.record.signal({
type: 'tool.call.started',
turnId: Number(event.turnId),
toolCallId: event.toolCallId,
@ -343,7 +341,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
return;
case 'tool.progress':
if (this.protocolTurnId === undefined) return;
this.events.emit({
this.record.signal({
type: 'tool.progress',
turnId: this.protocolTurnId,
toolCallId: event.toolCallId,
@ -353,7 +351,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
case 'tool.result':
if (this.protocolTurnId === undefined) return;
this.trackToolCallResult(event);
this.events.emit({
this.record.signal({
type: 'tool.result',
turnId: this.protocolTurnId,
toolCallId: event.toolCallId,

View file

@ -7,7 +7,7 @@ import {
type IDisposable,
} from "#/_base/di";
import { ErrorCodes, makeErrorPayload } from "#/errors";
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import { IAgentToolExecutorService } from '#/agent/toolExecutor';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import { createMcpAuthTool } from '#/agent/mcp/tools/auth';
@ -38,7 +38,7 @@ export class AgentMcpService extends Disposable implements IAgentMcpService {
constructor(
private readonly options: McpServiceOptions = {},
@IAgentToolRegistryService private readonly registry: IAgentToolRegistryService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentToolExecutorService toolExecutor: IAgentToolExecutorService,
) {
super();
@ -103,7 +103,7 @@ export class AgentMcpService extends Disposable implements IAgentMcpService {
}
private handleMcpServerStatusChange(entry: McpServerEntry): void {
this.events.emit({
this.record.signal({
type: 'mcp.server.status',
server: {
name: entry.name,
@ -123,7 +123,7 @@ export class AgentMcpService extends Disposable implements IAgentMcpService {
}
if (entry.status === 'failed') {
this.unregisterMcpServer(entry.name);
this.events.emit({
this.record.signal({
type: 'tool.list.updated',
reason: 'mcp.failed',
serverName: entry.name,
@ -133,7 +133,7 @@ export class AgentMcpService extends Disposable implements IAgentMcpService {
if (entry.status === 'disabled' || entry.status === 'pending') {
const removed = this.unregisterMcpServer(entry.name);
if (removed) {
this.events.emit({
this.record.signal({
type: 'tool.list.updated',
reason: 'mcp.disconnected',
serverName: entry.name,
@ -152,7 +152,7 @@ export class AgentMcpService extends Disposable implements IAgentMcpService {
resolved.enabledNames,
);
this.emitMcpToolCollisions(entry.name, result.collisions);
this.events.emit({
this.record.signal({
type: 'tool.list.updated',
reason: 'mcp.connected',
serverName: entry.name,
@ -173,7 +173,7 @@ export class AgentMcpService extends Disposable implements IAgentMcpService {
const disposable = this._register(this.registry.register(tool, { source: 'mcp' }));
this.mcpTools.set(tool.name, { disposable, serverName: entry.name });
this.mcpToolsByServer.set(entry.name, [tool.name]);
this.events.emit({
this.record.signal({
type: 'tool.list.updated',
reason: 'mcp.connected',
serverName: entry.name,
@ -251,7 +251,7 @@ export class AgentMcpService extends Disposable implements IAgentMcpService {
: `"${collision.toolName}" -> ${collision.qualified} (collides with server "${collision.collidesWith.serverName}")`,
)
.join('; ');
this.events.emit({
this.record.signal({
type: 'error',
...makeErrorPayload(
ErrorCodes.MCP_TOOL_NAME_COLLISION,

View file

@ -27,7 +27,7 @@ import { IAgentProfileService } from '#/agent/profile';
import { ITelemetryService } from '#/app/telemetry';
import { IAgentTurnService } from '#/agent/turn';
import type { ContextMessage } from '#/agent/contextMemory';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentRecordService } from '#/agent/record';
import {
IAgentMicroCompactionService,
type MicroCompactionConfig,
@ -66,7 +66,7 @@ export class AgentMicroCompactionService
constructor(
@IAgentContextMemoryService private readonly context: IAgentContextMemoryService,
@IAgentContextSizeService private readonly contextSize: IAgentContextSizeService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IFlagService private readonly flags: IFlagService,
@IAgentProfileService private readonly profile: IAgentProfileService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@ -93,13 +93,17 @@ export class AgentMicroCompactionService
),
);
this._register(
this.wireRecord.register('micro_compaction.apply', (record) => {
this.apply(record.cutoff);
this.record.define('micro_compaction.apply', {
resume: (r) => {
this.apply(r.cutoff);
},
}),
);
this._register(
this.wireRecord.register('full_compaction.complete', () => {
this.reset();
this.record.define('full_compaction.complete', {
resume: () => {
this.reset();
},
}),
);
this._register(
@ -119,7 +123,7 @@ export class AgentMicroCompactionService
}
private apply(cutoff: number): void {
this.wireRecord.append({
this.record.append({
type: 'micro_compaction.apply',
cutoff,
});
@ -210,7 +214,7 @@ export class AgentMicroCompactionService
}
if (context.messages.some(isAssistantCacheAnchor)) {
this._lastAssistantAt = this.wireRecord.restoring?.time ?? Date.now();
this._lastAssistantAt = this.record.restoring?.time ?? Date.now();
}
}

View file

@ -6,11 +6,8 @@ import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { IAgentContextInjectorService } from '#/agent/contextInjector';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { OrderedHookSlot } from '#/hooks';
import { IAgentReplayBuilderService } from '#/agent/replayBuilder';
import type { WireRecord } from '#/agent/wireRecord';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentRecordService } from '#/agent/record';
import { registerPermissionModeInjection } from '#/agent/permissionMode/injection/permissionModeInjection';
import { IAgentPermissionModeService } from './permissionMode';
@ -35,15 +32,17 @@ export class AgentPermissionModeService extends Disposable implements IAgentPerm
};
constructor(
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentContextInjectorService dynamicInjector: IAgentContextInjectorService,
) {
super();
this._register(
wireRecord.register('permission.set_mode', (record) => {
this.applyMode(record);
record.define('permission.set_mode', {
resume: (r) => {
this.applyMode(r.mode);
},
toLive: (r) => ({ type: 'agent.status.updated', permission: r.mode }),
toReplay: (r) => ({ type: 'permission_updated', mode: r.mode }),
}),
);
this._register(
@ -56,18 +55,13 @@ export class AgentPermissionModeService extends Disposable implements IAgentPerm
}
setMode(mode: PermissionMode): void {
this.wireRecord.append({ type: 'permission.set_mode', mode });
this.applyMode({ type: 'permission.set_mode', mode });
this.record.append({ type: 'permission.set_mode', mode });
this.applyMode(mode);
}
private applyMode(record: WireRecord<'permission.set_mode'>): void {
this.replayBuilder.push({ type: 'permission_updated', mode: record.mode });
private applyMode(mode: PermissionMode): void {
const previousMode = this.currentMode;
this.currentMode = record.mode;
this.events.emit({
type: 'agent.status.updated',
permission: this.currentMode,
});
this.currentMode = mode;
void this.hooks.onChanged.run({ mode: this.currentMode, previousMode });
}
}

View file

@ -15,13 +15,11 @@ import { toInputJsonSchema } from '#/_base/tools/support/input-schema';
import { generateHeroSlug } from "#/_base/utils/hero-slug";
import { IAgentContextMemoryService, type ContextMessage } from '#/agent/contextMemory';
import { IAgentContextInjectorService } from '#/agent/contextInjector';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { ISessionAgentFileSystem } from '#/session/agentFs';
import { IAgentProfileService } from '#/agent/profile';
import { IAgentReplayBuilderService } from '#/agent/replayBuilder';
import { ITelemetryService } from '#/app/telemetry';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentRecordService } from '#/agent/record';
import type { ToolInputDisplay } from '@moonshot-ai/protocol';
import type { ExecutableToolResult } from '#/agent/tool';
import { EnterPlanModeInputSchema } from '#/agent/plan/tools/enter-plan-mode';
@ -65,31 +63,36 @@ export class AgentPlanService extends Disposable implements IAgentPlanService {
constructor(
@IAgentContextMemoryService private readonly context: IAgentContextMemoryService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentRecordService private readonly record: IAgentRecordService,
@ISessionAgentFileSystem private readonly agentFs: ISessionAgentFileSystem,
@IAgentProfileService private readonly profile: IAgentProfileService,
@IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService,
@IAgentToolRegistryService toolRegistry: IAgentToolRegistryService,
@IAgentContextInjectorService dynamicInjector: IAgentContextInjectorService,
@ITelemetryService private readonly telemetry: ITelemetryService,
) {
super();
this._register(
wireRecord.register('plan_mode.enter', ({ id }) => {
this.restoreEnter({ id });
record.define('plan_mode.enter', {
resume: ({ id }) => {
this.restoreEnter({ id });
},
toReplay: () => ({ type: 'plan_updated', enabled: true }),
}),
);
this._register(
wireRecord.register('plan_mode.cancel', () => {
this.replayBuilder.push({ type: 'plan_updated', enabled: false });
this.applyInactive();
record.define('plan_mode.cancel', {
resume: () => {
this.applyInactive();
},
toReplay: () => ({ type: 'plan_updated', enabled: false }),
}),
);
this._register(
wireRecord.register('plan_mode.exit', () => {
this.replayBuilder.push({ type: 'plan_updated', enabled: false });
this.applyInactive();
record.define('plan_mode.exit', {
resume: () => {
this.applyInactive();
},
toReplay: () => ({ type: 'plan_updated', enabled: false }),
}),
);
@ -169,7 +172,7 @@ export class AgentPlanService extends Disposable implements IAgentPlanService {
const planFilePath = this.planFilePathFor(id);
this._planFilePath = planFilePath;
await this.ensurePlanDirectory(planFilePath);
this.wireRecord.append({ type: 'plan_mode.enter', id });
this.record.append({ type: 'plan_mode.enter', id });
enterRecorded = true;
if (createFile) {
await this.writeEmptyPlanFile(planFilePath);
@ -187,15 +190,13 @@ export class AgentPlanService extends Disposable implements IAgentPlanService {
}
private restoreEnter({ id }: { readonly id: string }): void {
this.replayBuilder.push({ type: 'plan_updated', enabled: true });
this._active = true;
this.planId = id;
this._planFilePath = this.planFilePathFor(id);
}
cancel(id?: string): void {
this.wireRecord.append({ type: 'plan_mode.cancel', id });
this.replayBuilder.push({ type: 'plan_updated', enabled: false });
this.record.append({ type: 'plan_mode.cancel', id });
this.applyInactive();
this.emitChanged();
}
@ -206,8 +207,7 @@ export class AgentPlanService extends Disposable implements IAgentPlanService {
}
exit(id?: string): void {
this.wireRecord.append({ type: 'plan_mode.exit', id });
this.replayBuilder.push({ type: 'plan_updated', enabled: false });
this.record.append({ type: 'plan_mode.exit', id });
this.applyInactive();
this.emitChanged();
}
@ -343,7 +343,7 @@ export class AgentPlanService extends Disposable implements IAgentPlanService {
}
private emitChanged(): void {
this.events.emit({ type: 'agent.status.updated', planMode: this._active });
this.record.signal({ type: 'agent.status.updated', planMode: this._active });
}
private trackTelemetry(

View file

@ -34,7 +34,7 @@ import { ISessionModelResolver, type ResolvedModel } from '#/session/modelRuntim
import { ISessionWorkspaceContext } from '#/session/workspaceContext';
import type { ResolvedAgentProfile, SystemPromptContext } from '#/agent/profile';
import { IAgentRecordService } from '#/agent/record';
import { IAgentRecordService, type AgentRecord } from '#/agent/record';
import { ITelemetryService } from '#/app/telemetry';
import type { ToolSource } from '#/agent/tool';
import { prepareSystemPromptContext } from './context';
@ -119,7 +119,7 @@ export class AgentProfileService implements IAgentProfileService {
update(changed: ProfileUpdateData): void {
const { activeToolNames, ...configChanged } = changed;
if (Object.keys(configChanged).length > 0) {
this.wireRecord.append({ type: 'config.update', ...configChanged });
this.record.append({ type: 'config.update', ...configChanged });
this.apply(configChanged);
}
if (activeToolNames !== undefined) {
@ -173,7 +173,7 @@ export class AgentProfileService implements IAgentProfileService {
const { agentsMdWarning } = context;
this.agentsMdWarning = agentsMdWarning;
if (agentsMdWarning !== undefined) {
this.events.emit({
this.record.signal({
type: 'warning',
message: agentsMdWarning,
code: 'agents-md-oversized',
@ -274,7 +274,6 @@ export class AgentProfileService implements IAgentProfileService {
}
private apply(changed: ProfileUpdateData): void {
this.replayBuilder.push({ type: 'config_updated', config: changed });
if (changed.cwd !== undefined) {
this.cwdValue = changed.cwd;
void this.optionsValue.chdir?.(changed.cwd);
@ -298,7 +297,7 @@ export class AgentProfileService implements IAgentProfileService {
}
private setActiveTools(names: readonly string[]): void {
this.wireRecord.append({ type: 'tools.set_active_tools', names: [...names] });
this.record.append({ type: 'tools.set_active_tools', names: [...names] });
this.applyActiveToolNames(names);
this.initializeBuiltinTools();
}
@ -319,7 +318,7 @@ export class AgentProfileService implements IAgentProfileService {
return;
}
if (!this.hasModel()) return;
this.events.emit({
this.record.signal({
type: 'agent.status.updated',
model: this.modelAlias,
maxContextTokens: this.getModelCapabilities().max_context_tokens,
@ -381,6 +380,11 @@ export class AgentProfileService implements IAgentProfileService {
}
}
function stripConfigMeta(record: AgentRecord<'config.update'>): ProfileUpdateData {
const { type: _type, time: _time, ...changed } = record;
return changed;
}
registerScopedService(
LifecycleScope.Agent,
IAgentProfileService,

View file

@ -8,9 +8,8 @@ import {
USER_PROMPT_ORIGIN,
type ContextMessage,
} from '#/agent/contextMemory';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import { IAgentTurnService, type Turn } from '#/agent/turn';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentPromptService } from './prompt';
export class AgentPromptService implements IAgentPromptService {
@ -21,8 +20,7 @@ export class AgentPromptService implements IAgentPromptService {
constructor(
@IAgentContextMemoryService private readonly context: IAgentContextMemoryService,
@IAgentTurnService private readonly turnService: IAgentTurnService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentRecordService private readonly record: IAgentRecordService,
) {
turnService.hooks.beforeStep.register('prompt-service-steer-before-step', async (_ctx, next) => {
this.flushSteerQueue();
@ -89,7 +87,7 @@ export class AgentPromptService implements IAgentPromptService {
}
}
if (!this.wireRecord.restoring && (removedUserCount < count || stoppedAtCompaction)) {
if (!this.record.restoring && (removedUserCount < count || stoppedAtCompaction)) {
throw new KimiError(
ErrorCodes.REQUEST_INVALID,
formatUndoUnavailableMessage(count, removedUserCount, stoppedAtCompaction),
@ -142,7 +140,7 @@ export class AgentPromptService implements IAgentPromptService {
private emitBusyIfActive(): boolean {
const activeTurn = this.turnService.getActiveTurn();
if (activeTurn === undefined) return false;
this.events.emit({
this.record.signal({
type: 'error',
...makeErrorPayload(
ErrorCodes.TURN_AGENT_BUSY,

View file

@ -0,0 +1,8 @@
/**
* `record` domain barrel re-exports the record service contract and
* implementation. Importing this barrel registers the `IAgentRecordService`
* binding into the scope registry.
*/
export * from './record';
export * from './recordService';

View file

@ -0,0 +1,106 @@
/**
* `record` domain (L3) `IAgentRecordService` contract.
*
* Single entry point for recording facts that happen inside an agent. One
* `append(record)` call fans out to every facet declared for the record type:
* durable persistence (for resume), live broadcast (`AgentEvent` to the edge),
* and replay capture. `signal(event)` emits a live-only event that is never
* recorded (deltas / progress). Bound at Agent scope.
*/
import type { AgentEvent } from '@moonshot-ai/protocol';
import type { IDisposable } from '#/_base/di';
import { createDecorator } from '#/_base/di';
import type {
IAgentWireRecordService,
WireRecord,
WireRecordBlobSelector,
WireRecordMap,
WireRecordRestoreOptions,
WireRecordRestoreResult,
PersistedWireRecord,
WireRecordRestoringContext,
} from '#/agent/wireRecord';
import type { AgentReplayRecord, AgentReplayRecordPayload } from '#/agent/replayBuilder/types';
/**
* Per-agent fact registry. Each domain augments `WireRecordMap` (in
* `#/agent/wireRecord`) to declare the payload shape of the records it owns;
* `AgentRecordMap` extends it so the same record types are accepted by the
* `record` facade and by the persistence layer (`PersistedWireRecord`) without
* duplication. New domains may augment either interface.
*/
export interface AgentRecordMap extends WireRecordMap {}
export type AgentRecord<K extends keyof AgentRecordMap = keyof AgentRecordMap> = {
[T in K]: { readonly type: T; readonly time?: number } & Readonly<AgentRecordMap[T]>;
}[K];
/**
* Facets declared for a record type. Every facet is optional: a record with no
* facets is still persisted (durable) but broadcasts nothing and captures no
* replay. Restore-time behavior (resume + replay capture) is driven entirely
* by these facets, so business code never re-implements the live/restore split.
*/
export interface RecordFacets<K extends keyof AgentRecordMap> {
/**
* Live projection: `record → AgentEvent` broadcast on the wire. When omitted
* the record is not broadcast live. Automatically suppressed while restoring.
*/
readonly toLive?: (record: AgentRecord<K>) => AgentEvent;
/**
* Replay projection: `record →` zero/one/many replay records. Runs on both
* the live `append` path and the restore path; the replay layer gates by
* phase so live appends are only captured when `captureLiveRecords` is set.
*/
readonly toReplay?: (
record: AgentRecord<K>,
) => AgentReplayRecordPayload | readonly AgentReplayRecordPayload[] | undefined;
/** Resumer: rebuild in-memory state from a restored record. */
readonly resume?: (record: AgentRecord<K>) => void | Promise<void>;
/**
* Blob offload/rehydrate selector for large content parts (e.g. images in
* context messages). Forwarded to the durable store so oversized parts are
* offloaded to the blob store on append and rehydrated on restore.
*/
readonly blobs?: WireRecordBlobSelector<AgentRecord<K>>;
}
export interface IAgentRecordService {
readonly _serviceBrand: undefined;
/** Record a fact: persists it and fans out to its declared facets. */
append(record: AgentRecord): void;
/**
* Subscribe to the live `AgentEvent` stream. Suppressed while restoring so
* edge consumers never receive historical events. This is the contract the
* `server-v2` edge depends on (`on(handler)` `AgentEvent`).
*/
on(handler: (event: AgentEvent) => void): IDisposable;
/** Emit a live-only event that is never persisted (deltas / progress). */
signal(event: AgentEvent): void;
/**
* Declare the facets for a record type. Call from a domain's constructor.
* Returns a disposable that unregisters the facets (and the resumer).
*/
define<K extends keyof AgentRecordMap>(type: K, facets: RecordFacets<K>): IDisposable;
restore(
records?: readonly PersistedWireRecord[],
options?: WireRecordRestoreOptions,
): Promise<WireRecordRestoreResult>;
flush(): Promise<void>;
close(): Promise<void>;
/** Replay result built from restored (and optionally live) records. */
buildReplay(): readonly AgentReplayRecord[];
/** When true, live `append` calls also feed the replay buffer. */
captureLiveRecords: boolean;
readonly restoring: WireRecordRestoringContext | null;
readonly postRestoring: boolean;
readonly hooks: IAgentWireRecordService['hooks'];
}
export const IAgentRecordService = createDecorator<IAgentRecordService>('agentRecordService');

View file

@ -0,0 +1,176 @@
/**
* `record` domain (L3) `IAgentRecordService` implementation.
*
* Owns the unified `append` / `signal` / `define` API: one `append(record)`
* fans out to durable persistence (delegated to `wireRecord`), live broadcast
* (an owned `Emitter<AgentEvent>`), and replay capture (delegated to
* `replayBuilder`). `signal(event)` emits a live-only event that is never
* recorded. Live emission is suppressed while restoring, so edge consumers
* never receive historical events. The former `eventSink` service is folded
* into this class; `wireRecord` / `replayBuilder` remain registered backends
* that this service coordinates.
*/
import { Disposable, toDisposable } from '#/_base/di';
import type { IDisposable } from '#/_base/di';
import { Emitter } from '#/_base/event';
import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import type { AgentEvent } from '@moonshot-ai/protocol';
import {
IAgentWireRecordService,
type WireRecord,
type WireRecordBlobSelector,
type WireRecordMap,
type WireRecordRestoreOptions,
type WireRecordRestoreResult,
type PersistedWireRecord,
type WireRecordRestoringContext,
} from '#/agent/wireRecord';
import { IAgentReplayBuilderService } from '#/agent/replayBuilder';
import type { AgentReplayRecord } from '#/agent/replayBuilder/types';
import {
IAgentRecordService,
type AgentRecord,
type AgentRecordMap,
type RecordFacets,
} from './record';
export class AgentRecordService extends Disposable implements IAgentRecordService {
declare readonly _serviceBrand: undefined;
private readonly facets = new Map<keyof AgentRecordMap, RecordFacets<keyof AgentRecordMap>>();
private readonly liveEmitter = this._register(new Emitter<AgentEvent>());
constructor(
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService,
) {
super();
// Restore-time replay capture: every restored record runs its `toReplay`
// facet. `replayBuilder.push` gates by phase, so this only records while
// restoring / post-restoring.
this._register(
wireRecord.hooks.onRestoredRecord.register('record-replay', async (ctx, next) => {
await next();
this.runReplayFacet(ctx.record as unknown as AgentRecord);
}),
);
}
append(record: AgentRecord): void {
this.wireRecord.append(record as unknown as WireRecord);
const facet = this.facets.get(record.type);
if (facet?.toLive !== undefined) {
this.emitLive(facet.toLive(record));
}
if (facet?.toReplay !== undefined) {
this.runReplayFacet(record);
}
}
on(handler: (event: AgentEvent) => void): IDisposable {
return this.liveEmitter.event(handler);
}
signal(event: AgentEvent): void {
this.emitLive(event);
}
define<K extends keyof AgentRecordMap>(type: K, facets: RecordFacets<K>): IDisposable {
// Merge live/replay facets rather than overwriting: the record type's owner
// supplies `toLive`/`toReplay`, while secondary resumers (other domains that
// listen to the same type, e.g. microCompaction listening to
// `full_compaction.complete`) only contribute a `resume`. First writer wins
// for live/replay; every `resume` is kept (the durable store supports
// multiple resumers per type).
const previous = this.facets.get(type);
this.facets.set(type, {
toLive: facets.toLive ?? previous?.toLive,
toReplay: facets.toReplay ?? previous?.toReplay,
} as RecordFacets<keyof AgentRecordMap>);
const resumeReg =
facets.resume === undefined
? undefined
: this.wireRecord.register(
type as unknown as keyof WireRecordMap,
(record) => facets.resume!(record as unknown as AgentRecord<K>),
facets.blobs === undefined
? undefined
: {
blobs: facets.blobs as unknown as WireRecordBlobSelector<
WireRecord<keyof WireRecordMap>
>,
},
);
return toDisposable(() => {
resumeReg?.dispose();
});
}
restore(
records?: readonly PersistedWireRecord[],
options?: WireRecordRestoreOptions,
): Promise<WireRecordRestoreResult> {
return this.wireRecord.restore(records, options);
}
flush(): Promise<void> {
return this.wireRecord.flush();
}
close(): Promise<void> {
return this.wireRecord.close();
}
buildReplay(): readonly AgentReplayRecord[] {
return this.replayBuilder.buildResult();
}
get captureLiveRecords(): boolean {
return this.replayBuilder.captureLiveRecords;
}
set captureLiveRecords(value: boolean) {
this.replayBuilder.captureLiveRecords = value;
}
get restoring(): WireRecordRestoringContext | null {
return this.wireRecord.restoring;
}
get postRestoring(): boolean {
return this.replayBuilder.postRestoring;
}
get hooks(): IAgentWireRecordService['hooks'] {
return this.wireRecord.hooks;
}
private emitLive(event: AgentEvent): void {
// Suppress live emission while restoring so edge consumers never receive
// historical events (matches the former `eventSink.emit` guard).
if (this.wireRecord.restoring !== null) return;
this.liveEmitter.fire(event);
}
private runReplayFacet(record: AgentRecord): void {
const facet = this.facets.get(record.type);
if (facet?.toReplay === undefined) return;
const out = facet.toReplay(record);
if (out === undefined) return;
const list = Array.isArray(out) ? out : [out];
for (const replayRecord of list) {
this.replayBuilder.push(replayRecord);
}
}
}
registerScopedService(
LifecycleScope.Agent,
IAgentRecordService,
AgentRecordService,
InstantiationType.Delayed,
'record',
);

View file

@ -8,7 +8,7 @@ import { IAgentContextSizeService } from '#/agent/contextSize';
import { IAgentFileToolsService } from '#/agent/fileTools';
import { IAgentFullCompactionService } from '#/agent/fullCompaction';
import { IAgentGoalService } from '#/agent/goal';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import { ErrorCodes, KimiError } from '#/errors';
import { userCancellationReason } from '#/_base/utils/abort';
import { IAgentPermissionGate } from '#/agent/permissionGate';
@ -96,7 +96,7 @@ export class AgentRPCService implements IAgentRPCService {
@IAgentUsageService private readonly usage: IAgentUsageService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@IAgentGoalService private readonly goal: IAgentGoalService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentQuestionToolsService private readonly questionTools: IAgentQuestionToolsService,
@IAgentWebService private readonly web: IAgentWebService,
@IPluginService private readonly plugins: IPluginService,
@ -149,12 +149,12 @@ export class AgentRPCService implements IAgentRPCService {
else if (update.kind === 'stderr') stderr += update.text ?? '';
else return;
if (payload.commandId !== undefined) {
this.events.emit({ type: 'shell.output', commandId: payload.commandId, update });
this.record.signal({ type: 'shell.output', commandId: payload.commandId, update });
}
},
onForegroundTaskStart: (taskId: string) => {
if (payload.commandId !== undefined) {
this.events.emit({ type: 'shell.started', commandId: payload.commandId, taskId });
this.record.signal({ type: 'shell.started', commandId: payload.commandId, taskId });
}
},
});
@ -316,7 +316,7 @@ export class AgentRPCService implements IAgentRPCService {
commandArgs: payload.args,
trigger: 'user-slash' as const,
};
this.events.emit({
this.record.signal({
type: 'plugin_command.activated',
activationId: origin.activationId,
pluginId: origin.pluginId,

View file

@ -19,12 +19,11 @@ import {
isUserActivatableSkillType,
type SkillDefinition,
} from '#/app/globalSkillCatalog/types';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentPromptService } from '#/agent/prompt';
import { ITelemetryService } from '#/app/telemetry';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import type { Turn } from '#/agent/turn';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentRecordService } from '#/agent/record';
import {
IAgentSkillService,
type ModelSkillActivationInput,
@ -47,15 +46,16 @@ export class AgentSkillService extends Disposable implements IAgentSkillService
constructor(
@ISessionSkillCatalog private readonly skillCatalog: ISessionSkillCatalog,
@IAgentPromptService private readonly prompt: IAgentPromptService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentRecordService private readonly records: IAgentRecordService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@IAgentToolRegistryService toolRegistry: IAgentToolRegistryService,
) {
super();
this._register(
this.wireRecord.register('skill.activate', (record) => {
this.publishActivation(record.origin);
records.define('skill.activate', {
resume: (r) => {
this.publishActivation(r.origin);
},
}),
);
this._register(toolRegistry.register(new SkillTool(this)));
@ -162,7 +162,7 @@ export class AgentSkillService extends Disposable implements IAgentSkillService
input?: readonly ContentPart[],
delivery: 'prompt' | 'steer' = 'prompt',
): Turn | undefined {
this.wireRecord.append({ type: 'skill.activate', origin });
this.records.append({ type: 'skill.activate', origin });
this.publishActivation(origin);
if (input === undefined) return undefined;
@ -180,7 +180,7 @@ export class AgentSkillService extends Disposable implements IAgentSkillService
}
private publishActivation(origin: SkillActivationOrigin): void {
this.events.emit({
this.records.signal({
type: 'skill.activated',
activationId: origin.activationId,
skillName: origin.skillName,
@ -189,7 +189,7 @@ export class AgentSkillService extends Disposable implements IAgentSkillService
skillPath: origin.skillPath,
skillSource: origin.skillSource,
});
if (this.wireRecord.restoring !== null) return;
if (this.records.restoring !== null) return;
this.telemetry.track('skill_invoked', {
skill_name: origin.skillName,
trigger: origin.trigger,

View file

@ -22,7 +22,7 @@ import {
} from '#/agent/agentTool';
import type { IAgentLifecycleService } from '#/session/agent-lifecycle';
import type { ISessionMetadata } from '#/session/session-metadata';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import { isUserCancellation } from '#/_base/utils/abort';
/*
@ -719,7 +719,7 @@ export function runChildAgentQueued<T>({
retryChildAgent({ lifecycle, parentAgentId, metadata, agentId, ...options }),
suspended: (event) => {
const parent = lifecycle.getHandle(parentAgentId);
parent?.accessor.get(IAgentEventSinkService)?.emit({
parent?.accessor.get(IAgentRecordService)?.signal({
type: 'subagent.suspended',
subagentId: event.agentId,
reason: event.reason,

View file

@ -11,12 +11,11 @@
import { Disposable } from '#/_base/di';
import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import { IAgentScopeContext } from '#/agent/scopeContext';
import { IAgentSystemReminderService } from '#/agent/systemReminder';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import { IAgentTurnService } from '#/agent/turn';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentLifecycleService } from '#/session/agent-lifecycle';
import SWARM_MODE_ENTER_REMINDER from './enter-reminder.md?raw';
import SWARM_MODE_EXIT_REMINDER from './exit-reminder.md?raw';
@ -42,8 +41,7 @@ export class AgentSwarmService extends Disposable implements IAgentSwarmService
constructor(
runQueued: AgentSwarmToolHost['runQueued'] | undefined,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentSystemReminderService private readonly reminders: IAgentSystemReminderService,
@IAgentTurnService turnService: IAgentTurnService,
@IAgentToolRegistryService toolRegistry: IAgentToolRegistryService,
@ -52,13 +50,17 @@ export class AgentSwarmService extends Disposable implements IAgentSwarmService
) {
super();
this._register(
wireRecord.register('swarm_mode.enter', (record) => {
this.restoreEnter(record.trigger);
record.define('swarm_mode.enter', {
resume: (r) => {
this.restoreEnter(r.trigger);
},
}),
);
this._register(
wireRecord.register('swarm_mode.exit', () => {
this.applyExit(false);
record.define('swarm_mode.exit', {
resume: () => {
this.applyExit(false);
},
}),
);
this._register(
@ -79,13 +81,13 @@ export class AgentSwarmService extends Disposable implements IAgentSwarmService
enter(trigger: SwarmModeTrigger): void {
if (this._active !== null) return;
this.wireRecord.append({ type: 'swarm_mode.enter', trigger });
this.record.append({ type: 'swarm_mode.enter', trigger });
this.applyEnter(trigger, true);
}
exit(): void {
if (this._active === null) return;
this.wireRecord.append({ type: 'swarm_mode.exit' });
this.record.append({ type: 'swarm_mode.exit' });
this.applyExit(true);
}
@ -124,7 +126,7 @@ export class AgentSwarmService extends Disposable implements IAgentSwarmService
}
private emitChanged(): void {
this.events.emit({ type: 'agent.status.updated', swarmMode: this.isActive });
this.record.signal({ type: 'agent.status.updated', swarmMode: this.isActive });
}
}

View file

@ -5,7 +5,7 @@ import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { OrderedHookSlot } from '#/hooks';
import { IAgentToolStoreService, type ToolStoreData, type ToolStoreKey } from './toolStore';
import { IAgentWireRecordService, type WireRecord } from '#/agent/wireRecord';
import { IAgentRecordService, type AgentRecord } from '#/agent/record';
declare module '#/agent/wireRecord' {
interface WireRecordMap {
@ -27,13 +27,13 @@ export class AgentToolStoreService extends Disposable implements IAgentToolStore
}>(),
};
constructor(
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
) {
constructor(@IAgentRecordService private readonly records: IAgentRecordService) {
super();
this._register(
wireRecord.register('tools.update_store', (record) => {
this.apply(record.key, record.value);
records.define('tools.update_store', {
resume: (r) => {
this.apply(r.key, r.value);
},
}),
);
}
@ -43,12 +43,12 @@ export class AgentToolStoreService extends Disposable implements IAgentToolStore
}
set<K extends ToolStoreKey>(key: K, value: ToolStoreData[K]): void {
const record: WireRecord<'tools.update_store'> = {
const record: AgentRecord<'tools.update_store'> = {
type: 'tools.update_store',
key,
value,
};
this.wireRecord.append(record);
this.records.append(record);
this.apply(key, value);
}
@ -58,7 +58,7 @@ export class AgentToolStoreService extends Disposable implements IAgentToolStore
private apply<K extends ToolStoreKey>(key: K, value: ToolStoreData[K]): void {
this.store[key] = value;
if (!this.wireRecord.restoring) {
if (!this.records.restoring) {
void this.hooks.onUpdated.run({ key, value });
}
}

View file

@ -4,12 +4,11 @@ import { toKimiErrorPayload, type KimiErrorPayload } from "#/errors";
import { isUserCancellation } from "#/_base/utils/abort";
import type { ContextMessage, PromptOrigin } from '#/agent/contextMemory';
import { IAgentContextMemoryService, USER_PROMPT_ORIGIN } from '#/agent/contextMemory';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentExternalHooksService } from '#/agent/externalHooks';
import { OrderedHookSlot } from '#/hooks';
import { IAgentLoopService } from '#/agent/loop';
import { ITelemetryService } from '#/app/telemetry';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentRecordService } from '#/agent/record';
import type {
Turn,
TurnContextOverflowContext,
@ -50,20 +49,21 @@ export class AgentTurnService implements IAgentTurnService {
constructor(
@IAgentLoopService private readonly loop: IAgentLoopService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentRecordService private readonly record: IAgentRecordService,
@IAgentContextMemoryService private readonly context: IAgentContextMemoryService,
@IAgentExternalHooksService private readonly externalHooks: IAgentExternalHooksService,
@ITelemetryService private readonly telemetry: ITelemetryService,
) {
wireRecord.register('turn.launch', (record) => {
this.restoreLaunch(record.turnId);
record.define('turn.launch', {
resume: (r) => {
this.restoreLaunch(r.turnId);
},
});
this.hooks.beforeStep.register('turn-before-step-event', async (ctx, next) => {
await next();
this.resolveReady(ctx.turn);
});
this.events.on((event) => {
this.record.on((event) => {
if (event.type === 'agent.status.updated' && event.planMode !== undefined) {
this.planModeActive = event.planMode;
return;
@ -86,7 +86,7 @@ export class AgentTurnService implements IAgentTurnService {
this.lastEndedReasonValue = undefined;
const turnId = this.nextTurnId;
this.wireRecord.append({ type: 'turn.launch', turnId, origin, promptMessageId });
this.record.append({ type: 'turn.launch', turnId, origin, promptMessageId });
this.restoreLaunch(turnId);
const abortController = new AbortController();
const ready = createControlledPromise<void>();
@ -120,7 +120,7 @@ export class AgentTurnService implements IAgentTurnService {
let result: TurnResult | undefined;
try {
this.telemetry.track('turn_started', { mode: telemetryMode });
this.events.emit({
this.record.signal({
type: 'turn.started',
turnId: turn.id,
origin,
@ -163,9 +163,9 @@ export class AgentTurnService implements IAgentTurnService {
) {
this.externalHooks.triggerInterrupt({ turnId: turn.id, reason: 'cancelled' });
}
this.events.emit(ended);
this.record.signal(ended);
if (ended.error !== undefined) {
this.events.emit({ type: 'error', ...ended.error });
this.record.signal({ type: 'error', ...ended.error });
}
if (ended.reason !== 'completed') {
this.trackTurnInterrupted(turn.id, 0);
@ -210,7 +210,7 @@ export class AgentTurnService implements IAgentTurnService {
toolCalls: [],
origin: { kind: 'hook_result', event: hookResult.event, blocked: true },
});
this.events.emit({
this.record.signal({
type: 'hook.result',
turnId: turn.id,
hookEvent: hookResult.event,
@ -227,7 +227,7 @@ export class AgentTurnService implements IAgentTurnService {
toolCalls: [],
origin: { kind: 'hook_result', event: hookResult.event },
});
this.events.emit({
this.record.signal({
type: 'hook.result',
turnId: turn.id,
hookEvent: hookResult.event,

View file

@ -5,10 +5,9 @@ import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { Disposable } from '#/_base/di/lifecycle';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import type { UsageRecordContext, UsageStatus } from './usage';
import { IAgentUsageService } from './usage';
import { IAgentWireRecordService } from '#/agent/wireRecord';
declare module '#/agent/wireRecord' {
interface WireRecordMap {
@ -26,20 +25,19 @@ export class AgentUsageService extends Disposable implements IAgentUsageService
private currentTurnId: number | undefined;
private currentTurn: TokenUsage | undefined;
constructor(
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentEventSinkService private readonly events: IAgentEventSinkService,
) {
constructor(@IAgentRecordService private readonly records: IAgentRecordService) {
super();
this._register(
wireRecord.register('usage.record', (record) => {
this.apply(record.model, record.usage, record.context);
records.define('usage.record', {
resume: (r) => {
this.apply(r.model, r.usage, r.context);
},
}),
);
}
record(model: string, usage: TokenUsage, context?: UsageRecordContext): void {
this.wireRecord.append({
this.records.append({
type: 'usage.record',
model,
usage,
@ -78,7 +76,7 @@ export class AgentUsageService extends Disposable implements IAgentUsageService
private publishChanged(): void {
const status = this.status();
if (status === undefined) return;
this.events.emit({ type: 'agent.status.updated', usage: status });
this.records.signal({ type: 'agent.status.updated', usage: status });
}
private byModelSnapshot(): Record<string, TokenUsage> {

View file

@ -12,7 +12,7 @@ import type {
import { ISessionInteractionService } from '#/session/interaction';
import { IAgentProfileService } from '#/agent/profile';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentRecordService } from '#/agent/record';
import {
IAgentUserToolService,
type UserToolRegistration,
@ -44,29 +44,33 @@ export class AgentUserToolService extends Disposable implements IAgentUserToolSe
constructor(
@IAgentToolRegistryService private readonly registry: IAgentToolRegistryService,
@IAgentProfileService private readonly profile: IAgentProfileService,
@IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService,
@IAgentRecordService private readonly records: IAgentRecordService,
@ISessionInteractionService private readonly interaction: ISessionInteractionService,
) {
super();
this._register(
wireRecord.register('tools.register_user_tool', (record) => {
this.applyRegister(record);
records.define('tools.register_user_tool', {
resume: (r) => {
this.applyRegister(r);
},
}),
);
this._register(
wireRecord.register('tools.unregister_user_tool', (record) => {
this.applyUnregister(record.name);
records.define('tools.unregister_user_tool', {
resume: (r) => {
this.applyUnregister(r.name);
},
}),
);
}
register(input: UserToolRegistration): void {
this.wireRecord.append({ type: 'tools.register_user_tool', ...input });
this.records.append({ type: 'tools.register_user_tool', ...input });
this.applyRegister(input);
}
unregister(name: string): void {
this.wireRecord.append({ type: 'tools.unregister_user_tool', name });
this.records.append({ type: 'tools.unregister_user_tool', name });
this.applyUnregister(name);
}

View file

@ -7,13 +7,13 @@
*
* WS event fan-out (sequencing, journaling, replay, per-connection dispatch)
* is a transport concern and lives in the edge package (`packages/server-v2`)
* on top of `IEventService` + `IAgentEventSinkService` not here.
* on top of `IEventService` + `IAgentRecordService` not here.
*/
import { InstantiationType } from '#/_base/di/extensions';
import { type IScopeHandle, LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { type IAgentScopeHandle, LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { IAgentLifecycleService } from '#/session/agent-lifecycle/agentLifecycle';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentRecordService } from '#/agent/record';
import { ILogService, ISessionLogService } from '#/app/log';
import { IAgentPromptService } from '#/agent/prompt';
import { ISessionLifecycleService } from '#/app/session-lifecycle';
@ -29,7 +29,7 @@ export class RestGateway implements IRestGateway {
@ILogService private readonly log: ILogService,
) {}
private agent(sessionId: string, agentId: string): IScopeHandle {
private agent(sessionId: string, agentId: string): IAgentScopeHandle {
const session = this.sessions.get(sessionId);
if (session === undefined) throw new Error(`unknown session '${sessionId}'`);
const agents = session.accessor.get(IAgentLifecycleService);
@ -92,7 +92,7 @@ export class WSGateway implements IWSGateway {
constructor(
@ISessionLifecycleService _sessions: ISessionLifecycleService,
@IAgentEventSinkService _event: IAgentEventSinkService,
@IAgentRecordService _record: IAgentRecordService,
) {}
connect(connectionId: string): void {

View file

@ -49,7 +49,6 @@ export * from '#/session/session-context';
export * from '#/session/session-activity';
export * from '#/session/session';
export * from '#/agent/eventSink';
import '#/session/approval';
export { ISessionApprovalService } from '#/session/approval';
export * from '#/session/question';
@ -76,7 +75,6 @@ export * from '#/agent/systemReminder';
export * from '#/agent/contextProjector';
export * from '#/agent/contextSize';
export * from '#/agent/contextInjector';
export * from '#/agent/eventSink';
export * from '#/agent/externalHooks';
export * from '#/agent/fullCompaction';
export * from '#/agent/llmRequestLog';
@ -92,9 +90,11 @@ export * from '#/agent/prompt';
export * from '#/agent/promptLegacy';
export * from '#/app/messageLegacy';
export * from '#/agent/replayBuilder';
export * from '#/agent/record';
export * from '#/agent/rpc';
export * from '#/agent/scopeContext';
export * from '#/agent/agentTool';
export * from '#/session/btw';
export * from '#/agent/todoList';
export * from '#/agent/tool';
export * from '#/agent/toolExecutor';

View file

@ -147,6 +147,7 @@ import type {
WireRecordRestoreOptions,
WireRecordRestoreResult,
} from '#/agent/wireRecord';
import { IAgentRecordService } from '#/agent/record';
import type { PathAccessOperation } from '#/session/workspaceContext';
import { createFakeAgentFs, createFakeHostEnvironment, createFakeProcessRunner } from '../tools/fixtures/fake-exec';
@ -213,12 +214,11 @@ class TestAgentSkillService extends AgentSkillService {
constructor(
@ISessionSkillCatalog skillCatalog: ISessionSkillCatalog,
@IAgentPromptService prompt: IAgentPromptService,
@IAgentEventSinkService events: IAgentEventSinkService,
@IAgentWireRecordService wireRecord: IAgentWireRecordService,
@IAgentRecordService records: IAgentRecordService,
@ITelemetryService telemetry: ITelemetryService,
@IAgentToolRegistryService toolRegistry: IAgentToolRegistryService,
) {
super(skillCatalog, prompt, events, wireRecord, telemetry, toolRegistry);
super(skillCatalog, prompt, records, telemetry, toolRegistry);
if (skillCatalog.catalog.listInvocableSkills().length > 0) {
this._register(toolRegistry.register(new ModelSkillTool(this)));
}

View file

@ -0,0 +1,207 @@
import { describe, expect, it, vi } from 'vitest';
import { DisposableStore, toDisposable } from '#/_base/di';
import { createServices } from '#/_base/di/test';
import { OrderedHookSlot } from '#/hooks';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentReplayBuilderService } from '#/agent/replayBuilder';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import type { WireRecord, WireRecordRestoredContext } from '#/agent/wireRecord';
import {
AgentRecordService,
IAgentRecordService,
type AgentRecord,
} from '#/agent/record';
import type { AgentEvent } from '@moonshot-ai/protocol';
import type { AgentReplayRecord, AgentReplayRecordPayload } from '#/agent/replayBuilder/types';
declare module '#/agent/record' {
interface AgentRecordMap {
'test.fact': { value: number };
}
}
interface StubHost {
readonly record: IAgentRecordService;
readonly wire: ReturnType<typeof createWireStub>;
readonly eventSink: ReturnType<typeof createEventSinkStub>;
readonly replay: ReturnType<typeof createReplayStub>;
readonly dispose: () => void;
}
function createWireStub() {
const appended: unknown[] = [];
const resumers = new Map<string, (record: unknown) => void | Promise<void>>();
const hooks = {
onRestoredRecord: new OrderedHookSlot<WireRecordRestoredContext>(),
onResumeEnded: new OrderedHookSlot<{}>(),
};
let restoring: { time?: number } | null = null;
return {
appended,
resumers,
hooks,
append: vi.fn((record: unknown) => appended.push(record)),
register: vi.fn((type: string, resumer: (record: unknown) => void | Promise<void>) => {
resumers.set(type, resumer);
return toDisposable(() => resumers.delete(type));
}),
restore: vi.fn(async () => ({}) as { warning?: string }),
flush: vi.fn(async () => {}),
close: vi.fn(async () => {}),
get restoring() {
return restoring;
},
setRestoring(value: { time?: number } | null) {
restoring = value;
},
postRestoring: false,
};
}
function createEventSinkStub() {
const emitted: AgentEvent[] = [];
return {
emitted,
emit: vi.fn((event: AgentEvent) => emitted.push(event)),
on: vi.fn(() => toDisposable(() => {})),
};
}
function createReplayStub() {
const records: AgentReplayRecord[] = [];
return {
records,
push: vi.fn((record: AgentReplayRecordPayload) =>
records.push(record as unknown as AgentReplayRecord),
),
buildResult: vi.fn(() => records),
captureLiveRecords: false,
postRestoring: false,
};
}
function createHost(): StubHost {
const wire = createWireStub();
const eventSink = createEventSinkStub();
const replay = createReplayStub();
const disposables = new DisposableStore();
const services = createServices(disposables, {
additionalServices: (reg) => {
reg.definePartialInstance(IAgentWireRecordService, wire);
reg.definePartialInstance(IAgentEventSinkService, eventSink);
reg.definePartialInstance(IAgentReplayBuilderService, replay);
reg.define(IAgentRecordService, AgentRecordService);
},
});
return {
record: services.get(IAgentRecordService),
wire,
eventSink,
replay,
dispose: () => disposables.dispose(),
};
}
describe('AgentRecordService facade', () => {
it('append fans out to durable + live + replay facets', () => {
const host = createHost();
host.record.define('test.fact', {
toLive: (r) => ({ type: 'test.live', value: r.value }) as unknown as AgentEvent,
toReplay: (r) =>
({ type: 'message', value: r.value }) as unknown as AgentReplayRecordPayload,
});
host.record.append({ type: 'test.fact', value: 42 });
expect(host.wire.append).toHaveBeenCalledWith(
expect.objectContaining({ type: 'test.fact', value: 42 }),
);
expect(host.eventSink.emitted).toContainEqual(
expect.objectContaining({ type: 'test.live', value: 42 }),
);
expect(host.replay.records).toContainEqual(
expect.objectContaining({ type: 'message', value: 42 }),
);
host.dispose();
});
it('append omits facets that are not declared', () => {
const host = createHost();
host.record.define('test.fact', {});
host.record.append({ type: 'test.fact', value: 1 });
expect(host.wire.append).toHaveBeenCalledTimes(1);
expect(host.eventSink.emit).not.toHaveBeenCalled();
expect(host.replay.push).not.toHaveBeenCalled();
host.dispose();
});
it('signal emits live only and never persists or captures replay', () => {
const host = createHost();
host.record.signal({ type: 'test.delta', delta: 'x' } as unknown as AgentEvent);
expect(host.eventSink.emitted).toContainEqual(
expect.objectContaining({ type: 'test.delta', delta: 'x' }),
);
expect(host.wire.append).not.toHaveBeenCalled();
expect(host.replay.push).not.toHaveBeenCalled();
host.dispose();
});
it('define registers the resumer with wireRecord and forwards the record', async () => {
const host = createHost();
const resume = vi.fn();
host.record.define('test.fact', { resume });
expect(host.wire.register).toHaveBeenCalledWith('test.fact', expect.any(Function));
const registered = host.wire.resumers.get('test.fact');
expect(registered).toBeDefined();
await registered?.({ type: 'test.fact', value: 7 });
expect(resume).toHaveBeenCalledWith(expect.objectContaining({ value: 7 }));
host.dispose();
});
it('captures replay during restore through the onRestoredRecord hook', async () => {
const host = createHost();
host.record.define('test.fact', {
toReplay: (r) =>
({ type: 'message', value: r.value }) as unknown as AgentReplayRecordPayload,
});
host.wire.setRestoring({ time: 123 });
await host.wire.hooks.onRestoredRecord.run({
record: { type: 'test.fact', value: 5 } as unknown as WireRecord,
stop: false,
});
expect(host.replay.records).toContainEqual(
expect.objectContaining({ type: 'message', value: 5 }),
);
host.dispose();
});
it('on() delegates to the live event sink', () => {
const host = createHost();
const handler = vi.fn();
host.record.on(handler);
expect(host.eventSink.on).toHaveBeenCalledWith(handler);
host.dispose();
});
it('dispose returned by define unregisters the resumer and facets', () => {
const host = createHost();
const subscription = host.record.define('test.fact', { resume: vi.fn() });
expect(host.wire.resumers.has('test.fact')).toBe(true);
subscription.dispose();
expect(host.wire.resumers.has('test.fact')).toBe(false);
// Facet removed: append should no longer fan out (still persists).
host.record.append({ type: 'test.fact', value: 1 });
expect(host.wire.append).toHaveBeenCalledTimes(1);
host.dispose();
});
});

View file

@ -20,6 +20,8 @@ import { ITelemetryService } from '#/app/telemetry';
import { IAgentToolRegistryService } from '#/agent/toolRegistry';
import type { Turn } from '#/agent/turn';
import { IAgentWireRecordService } from '#/agent/wireRecord';
import { IAgentReplayBuilderService } from '#/agent/replayBuilder';
import { AgentRecordService, IAgentRecordService } from '#/agent/record';
import { stubWireRecord } from '../contextMemory/stubs';
import { executeTool } from '../tools/fixtures/execute-tool';
import { stubSkill } from './stubs';
@ -71,6 +73,13 @@ describe('AgentSkillService', () => {
on: () => ({ dispose: () => {} }),
});
reg.defineInstance(IAgentWireRecordService, stubWireRecord());
reg.definePartialInstance(IAgentReplayBuilderService, {
push: () => {},
buildResult: () => [],
captureLiveRecords: false,
postRestoring: false,
});
reg.define(IAgentRecordService, AgentRecordService);
reg.definePartialInstance(ITelemetryService, { track: () => {} });
reg.definePartialInstance(IAgentToolRegistryService, {
register: () => ({ dispose: () => {} }),
@ -171,6 +180,13 @@ describe('SkillTool', () => {
on: () => ({ dispose: () => {} }),
});
reg.defineInstance(IAgentWireRecordService, stubWireRecord());
reg.definePartialInstance(IAgentReplayBuilderService, {
push: () => {},
buildResult: () => [],
captureLiveRecords: false,
postRestoring: false,
});
reg.define(IAgentRecordService, AgentRecordService);
reg.definePartialInstance(ITelemetryService, { track: () => {} });
reg.definePartialInstance(IAgentToolRegistryService, {
register: () => ({ dispose: () => {} }),

View file

@ -3,10 +3,14 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { toDisposable } from '#/_base/di';
import { DisposableStore } from '#/_base/di/lifecycle';
import { createServices, type TestInstantiationService } from '#/_base/di/test';
import { OrderedHookSlot } from '#/hooks';
import { IAgentEventSinkService } from '#/agent/eventSink';
import { IAgentReplayBuilderService } from '#/agent/replayBuilder';
import { AgentRecordService, IAgentRecordService } from '#/agent/record';
import { IAgentUsageService, type UsageStatus } from '#/agent/usage';
import { AgentUsageService } from '#/agent/usage/usageService';
import { IAgentWireRecordService, type WireRecord } from '#/agent/wireRecord';
import type { WireRecordRestoredContext } from '#/agent/wireRecord';
let disposables: DisposableStore;
@ -221,6 +225,10 @@ function createUsageHarness(): {
reg.definePartialInstance(IAgentWireRecordService, {
restoring: null,
postRestoring: false,
hooks: {
onRestoredRecord: new OrderedHookSlot<WireRecordRestoredContext>(),
onResumeEnded: new OrderedHookSlot<{}>(),
},
append: (record) => {
records.push(record);
},
@ -235,6 +243,13 @@ function createUsageHarness(): {
},
on: () => toDisposable(() => {}),
});
reg.definePartialInstance(IAgentReplayBuilderService, {
push: () => {},
buildResult: () => [],
captureLiveRecords: false,
postRestoring: false,
});
reg.define(IAgentRecordService, AgentRecordService);
reg.define(IAgentUsageService, AgentUsageService);
},
});

View file

@ -7,12 +7,12 @@
* Core `events` process-wide `DomainEvent` bus (`IEventService`)
* Session `interactions` pending human-in-the-loop requests (`ISessionInteractionService.onDidChange`)
* Session `interactions:resolved` request resolutions (`ISessionInteractionService.onDidResolve`)
* Agent `events` per-agent `AgentEvent` stream (`IAgentEventSinkService`)
* Agent `events` per-agent `AgentEvent` stream (`IAgentRecordService`)
*/
import {
IEventService,
IAgentEventSinkService,
IAgentRecordService,
ISessionInteractionService,
type DomainEvent,
type IDisposable,
@ -58,7 +58,7 @@ export const eventMap: Record<ScopeKind, Record<string, EventSource>> = {
},
agent: {
events: {
subscribe: (scope, listener) => scope.accessor.get(IAgentEventSinkService).on(listener),
subscribe: (scope, listener) => scope.accessor.get(IAgentRecordService).on(listener),
},
},
};

View file

@ -1,13 +1,13 @@
/**
* `SessionEventBroadcaster` per-session single fan-out point that turns
* agent `IAgentEventSinkService` emissions into a sequenced, journaled, replayable
* agent `IAgentRecordService` emissions into a sequenced, journaled, replayable
* `/api/v1/ws` event stream (the `{seq, epoch}` watermark).
*
* Port of v1's `WSBroadcastService` (`packages/server/.../wsBroadcastService.ts`),
* adapted to v2 where agent events live on per-agent `IAgentEventSinkService`s (not a Core
* adapted to v2 where agent events live on per-agent `IAgentRecordService`s (not a Core
* firehose). For each session it:
*
* 1. Subscribes to every agent's `IAgentEventSinkService` via `IAgentLifecycleService`
* 1. Subscribes to every agent's `IAgentRecordService` via `IAgentLifecycleService`
* reach-down-via-handle (and `onDidCreate`/`onDidDispose` for late agents).
* 2. Attaches `agentId`/`sessionId` to build the wire `Event`.
* 3. Classifies durable vs volatile (`VOLATILE_EVENT_TYPES`).
@ -24,10 +24,16 @@
* the journal is continuous from first activation onward.
*/
import type { DomainEvent, IDisposable, IScopeHandle, Scope } from '@moonshot-ai/agent-core-v2';
import type {
DomainEvent,
IAgentScopeHandle,
IDisposable,
ISessionScopeHandle,
Scope,
} from '@moonshot-ai/agent-core-v2';
import {
IAgentLifecycleService,
IAgentEventSinkService,
IAgentRecordService,
IEventService,
ISessionLifecycleService,
} from '@moonshot-ai/agent-core-v2';
@ -247,11 +253,11 @@ export class SessionEventBroadcaster {
state.queue = state.queue.then(() => this.dispatch(state, event)).catch(() => {});
}
private attachAgents(sessionId: string, session: IScopeHandle, state: SessionState): void {
private attachAgents(sessionId: string, session: ISessionScopeHandle, state: SessionState): void {
const agents = session.accessor.get(IAgentLifecycleService);
const subscribeAgent = (handle: IScopeHandle): void => {
const subscribeAgent = (handle: IAgentScopeHandle): void => {
if (state.agentDisposables.has(handle.id)) return;
const sink = handle.accessor.get(IAgentEventSinkService);
const sink = handle.accessor.get(IAgentRecordService);
const d = sink.on((agentEvent) => this.onAgentEvent(sessionId, handle.id, agentEvent));
state.agentDisposables.set(handle.id, d);
};