refactor(subagent): narrow continuation interface

This commit is contained in:
Tianyi Cui
2026-07-28 00:10:24 +08:00
committed by imccyu
parent 644bf00b86
commit f14121a4c2
55 changed files with 669 additions and 441 deletions

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/core/session/README.md
README.md: 59e8694a957e9742a22662766d671dc2145c44e3
README.zh.md: 7618bc8f3a9146a4fc5afbfb19317deef7f13068
README.md: 4730cac913e949d642d049a5c53ab2dd47e10627
README.zh.md: 12aa1625d7b0568196cd788c2a25cfb869d8780d

View File

@@ -13,8 +13,7 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall
### Public API
- `ctx.sessions.create(id?, { seed?, meta? }?)` validates and detaches durable seed/header data, fills the version and id, defaults `createdAt` to now, publishes the session, and binds it to the calling fiber. Persisted reconstruction supplies its original `createdAt`, `seedLength`, and `delegationDepth`.
- `ctx.sessions.flush(session)` dispatches the awaited parallel durability checkpoint through the session's captured scope. Every listener starts and the call waits for all to settle before reporting failure; unpublished, detached, and stale objects reject.
- `ctx.sessions.flushRequired(session)` uses the same dispatch but also rejects an empty scoped listener snapshot. Callers use it when success requires an installed durability participant rather than optional best-effort persistence.
- `ctx.sessions.flush(session)` dispatches the awaited parallel durability checkpoint through the session's captured scope. Every listener starts and the call waits for all to settle before reporting failure; it returns `true` when at least one listener participated and `false` for an empty snapshot, while unpublished, detached, and stale objects reject. A caller that requires durable storage rejects `false` at its own policy boundary.
- `findLastMessageTurnEnd(events)` pairs message-triggered starts with their ends and returns the latest matched `turn/end`. Outcome consumers use this fold instead of the raw latest log event because between-turn records and non-message turns have no prompt outcome.
- `ctx.sessions.fork(source, boundary?, childSessionId?): Session` — Resolve a live session object or id, select a seed through the inclusive `boundary` event seq (default: current last event), require that prefix to end outside an open turn, and create a live child session with lineage metadata.
- `ctx.sessions.get(id: SessionId): Session | undefined`

View File

@@ -13,8 +13,7 @@
### 公共 API
- `ctx.sessions.create(id?, { seed?, meta? }?)` 校验持久种子/头部数据并生成脱离副本,补齐版本和 id在未提供 `createdAt` 时使用当前时间,发布会话并将其绑定到调用方 fiber。持久化重建会提供原始的 `createdAt``seedLength``delegationDepth`
- `ctx.sessions.flush(session)` 通过会话捕获的作用域分发受等待的并行持久性检查点。每个监听器都会启动;调用会等待全部结算后才报告失败。未发布、已脱离和陈旧的对象会被拒绝
- `ctx.sessions.flushRequired(session)` 沿用相同的分发逻辑,但也会拒绝空的作用域监听器快照。若成功要求已安装的持久性参与方介入,而不是采用可选的尽力持久化,调用方应使用此方法。
- `ctx.sessions.flush(session)` 通过会话捕获的作用域分发受等待的并行持久性检查点。每个监听器都会启动;调用会等待全部结算后才报告失败;至少一个监听器参与时返回 `true`,监听器快照为空时返回 `false`,而未发布、已脱离和陈旧的对象会被拒绝。要求持久化存储的调用方应在自己的策略边界拒绝 `false`
- `findLastMessageTurnEnd(events)` 将由消息触发的开始与结束配对,并返回最近匹配的 `turn/end`。结果消费方使用该折叠逻辑,而不直接取日志中最近的事件,因为轮次间记录和非消息轮次没有提示词结果。
- `ctx.sessions.fork(source, boundary?, childSessionId?): Session`:解析实时会话对象或 id选取截至 `boundary` 事件序号(含该事件)的种子(默认为当前最后一个事件),要求所选前缀结束时没有开放轮次,再创建带谱系元数据的实时子会话。
- `ctx.sessions.get(id: SessionId): Session | undefined`

View File

@@ -93,9 +93,7 @@ declare module 'cordis' {
'session/event'(this: Scoped<Session>, session: Session, event: SessionEvent): void
/**
* Awaited parallel durability checkpoint: every listener runs and the
* caller awaits all of them, with no waterfall veto. An empty listener
* snapshot is accepted by {@link SessionStore.flush} and rejected by
* {@link SessionStore.flushRequired}. Scope-filtered dispatch
* caller awaits all of them, with no waterfall veto. Scope-filtered dispatch
* (`@deepseek-ai/dsh-scope`) reuses the session's owner scope.
* @param session - the session whose buffered events must reach durable storage.
* @dshScopeScan unsupported
@@ -970,35 +968,14 @@ export class SessionStore extends Service {
* raw `ctx.parallel('session/flush', …)` — one owner, one spelling, and the
* scoped-dispatch invariant can pin it.
* @param session - the session whose buffered events must reach durable storage.
* @returns resolves when every flush listener has settled; after all settle,
* rejects with the first registered listener failure if any listener failed.
* @returns whether at least one durability listener participated, after every
* listener has settled successfully.
* @throws the first registered listener failure after every listener settles.
*/
async flush(session: Session): Promise<void> {
await this.dispatchFlush(session, false)
}
/**
* Dispatch the same awaited checkpoint as {@link flush}, but reject when its
* scoped listener snapshot is empty. Callers use this operation when success
* requires an installed durability participant rather than optional
* best-effort persistence.
* @param session - the session whose buffered events must reach durable storage.
* @returns resolves when at least one listener participated and every
* listener settled successfully.
* @throws when no listener is registered or any registered listener fails.
*/
async flushRequired(session: Session): Promise<void> {
await this.dispatchFlush(session, true)
}
/** Dispatch one optional or required flush listener snapshot. */
private async dispatchFlush(session: Session, requireListener: boolean): Promise<void> {
async flush(session: Session): Promise<boolean> {
const { carrier } = this.liveEntryFor(session)
const callbackArgs: unknown[] = [session]
const callbacks = collectSessionCallbacks(this.ctx, [carrier, 'session/flush', session])
if (requireListener && callbacks.length === 0) {
throw new Error(`session "${session.id}" required durability checkpoint has no registered listener`)
}
const results = await Promise.allSettled(callbacks.map((callback) => {
try {
return callback(...callbackArgs)
@@ -1011,6 +988,7 @@ export class SessionStore extends Service {
}))
const failure = results.find((result): result is PromiseRejectedResult => result.status === 'rejected')
if (failure !== undefined) throw failure.reason
return callbacks.length > 0
}
/** Return the exact live entry; detached/prepared objects reject. */

View File

@@ -84,25 +84,16 @@ describe('sessions.flush()', () => {
const ctx = await mount()
const session = ctx.sessions.create()
await expect(ctx.sessions.flush(session)).resolves.toBeUndefined()
await expect(ctx.sessions.flush(session)).resolves.toBe(false)
})
it('rejects a required flush with no listeners', async () => {
const ctx = await mount()
const session = ctx.sessions.create()
await expect(ctx.sessions.flushRequired(session)).rejects.toThrow(
`session "${session.id}" required durability checkpoint has no registered listener`,
)
})
it('completes a required flush when a listener succeeds', async () => {
it('reports a participating listener after it succeeds', async () => {
const ctx = await mount()
const session = ctx.sessions.create()
const flushed: Session[] = []
ctx.on('session/flush', current => void flushed.push(current))
await ctx.sessions.flushRequired(session)
await expect(ctx.sessions.flush(session)).resolves.toBe(true)
expect(flushed).toEqual([session])
})