From 6def839f56564af6709e6dc1db5e4a229646a520 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Fri, 21 Aug 2026 16:12:35 +0800 Subject: [PATCH] perf(team): project durable state incrementally --- ...ession-projection-mandatory-seam.i18n.yaml | 4 +- ...08-19-session-projection-mandatory-seam.md | 2 +- ...19-session-projection-mandatory-seam.zh.md | 2 +- .../feature/2026-08-05-agent-teams.i18n.yaml | 4 +- .../feature/2026-08-05-agent-teams.md | 6 +- .../feature/2026-08-05-agent-teams.zh.md | 6 +- packages/experimental/team/README.i18n.yaml | 4 +- packages/experimental/team/README.md | 6 +- packages/experimental/team/README.zh.md | 6 +- packages/experimental/team/package.json | 1 + packages/experimental/team/src/index.ts | 9 +- packages/experimental/team/src/invariant.ts | 24 +-- packages/experimental/team/src/journal.ts | 17 +- packages/experimental/team/src/mailbox.ts | 14 +- .../team/src/{fold.ts => projection.ts} | 160 +++++++++++------- packages/experimental/team/src/roster.ts | 28 +-- packages/experimental/team/src/task-board.ts | 32 ++-- packages/experimental/team/src/task-graph.ts | 4 +- .../experimental/team/tests/invariant.spec.ts | 70 -------- .../team/tests/persistence.spec.ts | 17 +- ...fold.spec.ts => projection-events.spec.ts} | 107 +++++++----- packages/experimental/team/tests/team.spec.ts | 17 +- packages/experimental/team/tsconfig.json | 1 + packages/experimental/team/tsdown.config.ts | 2 +- .../session/session-projection/src/index.ts | 2 +- .../session-projection/tests/registry.spec.ts | 17 ++ 26 files changed, 282 insertions(+), 280 deletions(-) rename packages/experimental/team/src/{fold.ts => projection.ts} (65%) delete mode 100644 packages/experimental/team/tests/invariant.spec.ts rename packages/experimental/team/tests/{fold.spec.ts => projection-events.spec.ts} (70%) diff --git a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.i18n.yaml b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.i18n.yaml index 5a3db3ae1f..31a3631ce7 100644 --- a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.i18n.yaml @@ -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 .agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.md -2026-08-19-session-projection-mandatory-seam.md: 9edbece131cdc014f2cdf2ed0199b492b46b3ed1 -2026-08-19-session-projection-mandatory-seam.zh.md: e49081759682315f3503ee3114bebbf60c2d82e9 +2026-08-19-session-projection-mandatory-seam.md: eb76ac90fa760108cbeff1033c8ec2db71c7fa11 +2026-08-19-session-projection-mandatory-seam.zh.md: d0fcb2d903b08b71d8f4c9af62d3739c9ec6970b diff --git a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.md b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.md index 9edbece131..eb76ac90fa 100644 --- a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.md +++ b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.md @@ -16,7 +16,7 @@ This decision builds on the split between host projection state and client views The registry provides `stateOf(session, key)` for one typed host state and keeps `snapshot()` for batch carriers. Client views contain only consumed fields; host readers use `stateOf` for richer state. -`onChanged` publishes client-visible value changes only. Unit registration and removal remain effect-scoped registry lifecycle; they do not create a second Host event stream or client tombstone protocol. A later authoritative history or list baseline reflects the active key set. +`onChanged` publishes client-visible value changes only. Unit registration and removal remain effect-scoped registry lifecycle; `register()` returns the exact Cordis disposer so a composite domain owner can finish cleanup against projected state before removing its unit. Registration changes do not create a second Host event stream or client tombstone protocol. A later authoritative history or list baseline reflects the active key set. ## Alternatives considered diff --git a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.zh.md b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.zh.md index e490817596..d0fcb2d903 100644 --- a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.zh.md +++ b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.zh.md @@ -16,7 +16,7 @@ Status: implemented 注册表提供 `stateOf(session, key)` 来读取一个类型化 host 状态，并为批量 carrier 保留 `snapshot()`。客户端 view 只包含消费方使用的字段；host 读取方通过 `stateOf` 取得更丰富的状态。 -`onChanged` 只发布客户端可见值的变化。单元注册和移除仍是绑定 effect 的注册表生命周期；它们不创建第二条 Host 事件流或客户端 tombstone 协议。后续权威 history 或 list 基线会反映活跃 key 集。 +`onChanged` 只发布客户端可见值的变化。单元注册和移除仍是绑定 effect 的注册表生命周期；`register()` 返回 Cordis 的原始 disposer，使组合式领域 owner 可以先依据投影状态完成清理，再移除自身单元。注册变化不会创建第二条 Host 事件流或客户端 tombstone 协议。后续权威 history 或 list 基线会反映活跃 key 集。 ## 考虑过的替代方案 diff --git a/.agents/notes/implemented/feature/2026-08-05-agent-teams.i18n.yaml b/.agents/notes/implemented/feature/2026-08-05-agent-teams.i18n.yaml index f5ec3193a1..41bb2117be 100644 --- a/.agents/notes/implemented/feature/2026-08-05-agent-teams.i18n.yaml +++ b/.agents/notes/implemented/feature/2026-08-05-agent-teams.i18n.yaml @@ -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 .agents/notes/implemented/feature/2026-08-05-agent-teams.md -2026-08-05-agent-teams.md: 43e92d4a9807e6a178436d96722b141b1bac45bb -2026-08-05-agent-teams.zh.md: 72989623bb49334beae3b49ce2f6973781558c69 +2026-08-05-agent-teams.md: 51518f5b3a9ef61ad103432793754adabfb30f65 +2026-08-05-agent-teams.zh.md: 87ad12db08076b28ceb2cea4fafbf3ac09182bd7 diff --git a/.agents/notes/implemented/feature/2026-08-05-agent-teams.md b/.agents/notes/implemented/feature/2026-08-05-agent-teams.md index 43e92d4a98..51518f5b3a 100644 --- a/.agents/notes/implemented/feature/2026-08-05-agent-teams.md +++ b/.agents/notes/implemented/feature/2026-08-05-agent-teams.md @@ -14,7 +14,7 @@ The model-visible Team tools remain opt-in so the default tool catalog and simpl ## Decision -Every ordinary runtime root is the implicit Lead of a Team identified by that root's `SessionId`. The Team has no creation event: its Lead pseudo-row exists by identity, while durable state begins with the first member, message, or task event. A roster is flat and contains at most the configured number of immutable lowercase-kebab-case names. Each teammate is a continuable direct child with a reserved Session id; only the Lead creates or interrupts teammates. Ordinary provider-owned subagents outside the roster are not Team members, and an ordinary fork is a new root whose inherited Team records are excluded by their ancestor `TeamId`. +Every ordinary runtime root is the implicit Lead of a Team identified by that root's `SessionId`. The Team has no creation event: its Lead pseudo-row exists by identity, while durable state begins with the first member, message, or task event. A host-only Session projection eagerly folds committed Team records into checkpoint-safe JSON grouped by `TeamId`; host reads select the current Lead's group directly without replaying the full Session log. Grouping retains ancestor records for faithful checkpoint recovery while excluding them from an ordinary fork's new root state. A roster is flat and contains at most the configured number of immutable lowercase-kebab-case names. Each teammate is a continuable direct child with a reserved Session id; only the Lead creates or interrupts teammates. Ordinary provider-owned subagents outside the roster are not Team members. The implementation is split into `@deepseek-ai/dsh-team`, which owns `ctx.teams` and durable semantics, and `@deepseek-ai/dsh-tool-team`, which owns scoped schemas and model guidance. Every Team tool declares its complete result schema and renders that value as compact JSON, so the compiler checks each `execute` against what the model is promised and no result spends tokens on indentation. Deployments mount both plugins explicitly and may disable legacy continuable controls with the same model-visible names. The explicit delegation policy permits Team creation only when the user asks for Agent Teams or teammates. Both packages are private members of `packages/experimental/`; the [experimental package decision](../architecture/2026-08-18-experimental-agent-teams-packages.md) owns release exclusion, dependency isolation, and promotion. @@ -30,7 +30,7 @@ Fresh children have no inherited conversation. Fork children capture the Lead's ## Mailbox and task transactions -Peer communication is a Lead-log mailbox. `team/message/queued` is appended and flushed before delivery. The target message carries the stable message id and sender identity in both durable source metadata and a short model-visible prefix. A target receipt is acknowledged with `team/message/delivered` only after its pending inbox item or recorded user message is flushed. Immediate admission is serialized per target in queued-log order, recovery retries queued-minus-delivered in the same order, and delivery folds live or persisted target inbox/history state before cold resume. Every current-version Team payload is runtime-validated before entering replay state. The Team runtime tracks dispatch and asynchronous acknowledgement work from synchronous admission until settlement; disposal closes admission and awaits both before removing the service. Current waiters wake only after the owning Team event flush succeeds. +Peer communication is a Lead-log mailbox. `team/message/queued` is appended and flushed before delivery. The target message carries the stable message id and sender identity in both durable source metadata and a short model-visible prefix. A target receipt is acknowledged with `team/message/delivered` only after its pending inbox item or recorded user message is flushed. Immediate admission is serialized per target in queued-log order, recovery retries queued-minus-delivered in the same order, and delivery folds live or persisted target inbox/history state before cold resume. Every current-version Team payload is runtime-validated before entering projected state. The Team runtime tracks dispatch and asynchronous acknowledgement work from synchronous admission until settlement; disposal closes admission and awaits both before removing the service. Current waiters wake only after the owning Team event flush succeeds. Quiet `send_message` injects, flushes, and acknowledges immediately for a live target without waking it; an inactive target remains queued until another event materializes that teammate. Waking `followup_task` becomes the target's next FIFO turn and may cold-resume it. Success means the message is already durable even when immediate delivery is deferred. The mechanism provides process-local retry and target-Session de-duplication, not a cross-process exactly-once claim. @@ -62,7 +62,7 @@ Worktree isolation is not a harness runtime behavior. A deployment or prompt may ## Testing -Package tests cover identity, name and authority checks, provider selection, reserved-id persistence collisions, child-before-Lead flush ordering, durable provisioning failure and pending-inbox JSONL/SQLite reconciliation, concurrent target-local ordering, pending/history de-duplication, mailbox limits, post-flush notification, bounded disposal with in-flight creation and dispatch cancellation, failed-member cleanup, task CAS and DAG validation, write-scope warnings, wait cancellation/timeout, inbox-preserving interruption, ordinary-fork isolation, legacy-control shadowing, compact declared-schema result rendering, and scoped registration HMR at per-file 100% coverage. A keyless headless Loader snapshot assembles the real Team plugins and records teammate creation, peer mail, dependent tasks, waiting, and Lead aggregation. +Package tests cover identity, name and authority checks, provider selection, reserved-id persistence collisions, child-before-Lead flush ordering, durable provisioning failure and pending-inbox JSONL/SQLite reconciliation, concurrent target-local ordering, pending/history de-duplication, mailbox limits, post-flush notification, bounded disposal with in-flight creation and dispatch cancellation, failed-member cleanup, task CAS and DAG validation, write-scope warnings, wait cancellation/timeout, inbox-preserving interruption, Team projection checkpoint state, ordinary-fork isolation, legacy-control shadowing, compact declared-schema result rendering, and scoped registration HMR at per-file 100% coverage. A keyless headless Loader snapshot assembles the real Team plugins and records teammate creation, peer mail, dependent tasks, waiting, and Lead aggregation. ## Consequences diff --git a/.agents/notes/implemented/feature/2026-08-05-agent-teams.zh.md b/.agents/notes/implemented/feature/2026-08-05-agent-teams.zh.md index 72989623bb..87ad12db08 100644 --- a/.agents/notes/implemented/feature/2026-08-05-agent-teams.zh.md +++ b/.agents/notes/implemented/feature/2026-08-05-agent-teams.zh.md @@ -14,7 +14,7 @@ subagent seam 已提供 fresh／fork provider、持久 child Session、FIFO foll ## Decision -每个普通运行时 Root 都是一个隐式 Team 的 Lead，Team id 等于该 Root 的 `SessionId`。Team 没有 creation event：Lead pseudo-row 由身份直接存在，持久状态从第一条 member、message 或 task event 开始。roster 是扁平结构，最多包含配置数量、不可变且采用小写 kebab-case 的名字。每个 teammate 都是使用预留 Session id 的 continuable 直接 child；只有 Lead 可以创建或 interrupt teammate。roster 外由 provider 管理的普通 subagent 不是 Team member；普通 fork 是新的 Root，继承的 Team 记录会因 ancestor `TeamId` 被排除。 +每个普通运行时 Root 都是一个隐式 Team 的 Lead，Team id 等于该 Root 的 `SessionId`。Team 没有 creation event：Lead pseudo-row 由身份直接存在，持久状态从第一条 member、message 或 task event 开始。host-only Session projection 会把已提交 Team 记录即时折叠成按 `TeamId` 分组且可安全 checkpoint 的 JSON；host 读取会直接选择当前 Lead 的分组，无需重放完整 Session 日志。分组会保留 ancestor 记录以支持准确的 checkpoint 恢复，同时把它们排除在普通 fork 的新 Root 状态之外。roster 是扁平结构，最多包含配置数量、不可变且采用小写 kebab-case 的名字。每个 teammate 都是使用预留 Session id 的 continuable 直接 child；只有 Lead 可以创建或 interrupt teammate。roster 外由 provider 管理的普通 subagent 不是 Team member。 实现拆分为 `@deepseek-ai/dsh-team` 与 `@deepseek-ai/dsh-tool-team`：前者负责 `ctx.teams` 和持久语义，后者负责 scoped schema 与模型指引。每个 Team 工具都声明完整的结果 schema，并把该值渲染为紧凑 JSON，因此编译器会检查每个 `execute` 是否符合对模型的承诺，也没有结果把 token 花在缩进上。部署显式挂载两个插件，并可禁用具有相同模型可见名称的旧 continuable control。显式 delegation 策略只允许在用户要求 Agent Teams 或 teammate 时创建 Team。 两个包都是 `packages/experimental/` 的私有成员；[实验性包决策](../architecture/2026-08-18-experimental-agent-teams-packages.md)负责发布排除、依赖隔离与 promotion。 @@ -30,7 +30,7 @@ fresh child 不继承对话。fork child 只捕获一次 Lead 已完成 turn 前 ## Mailbox and task transactions -Peer 通讯使用 Lead 日志 mailbox。投递前先追加并 flush `team/message/queued`。target message 会在持久 source metadata 与短模型可见前缀中同时携带稳定 message id 和 sender identity。只有 pending inbox 条目或已记录用户消息完成 flush，Lead 日志才写入 `team/message/delivered` acknowledgement。即时准入按 target 和 queued 日志顺序串行化，恢复按同一顺序重试 queued-minus-delivered，并在冷恢复前折叠 live 或 persisted target 的 inbox／历史状态。每个当前版本 Team payload 都会经过运行时验证后才进入 replay state。Team runtime 从同步准入到 settlement 全程跟踪 dispatch 与异步 acknowledgement 工作；dispose 会关闭准入，并在移除服务前等待两者。当前 waiter 只在所属 Team event flush 成功后被唤醒。 +Peer 通讯使用 Lead 日志 mailbox。投递前先追加并 flush `team/message/queued`。target message 会在持久 source metadata 与短模型可见前缀中同时携带稳定 message id 和 sender identity。只有 pending inbox 条目或已记录用户消息完成 flush，Lead 日志才写入 `team/message/delivered` acknowledgement。即时准入按 target 和 queued 日志顺序串行化，恢复按同一顺序重试 queued-minus-delivered，并在冷恢复前折叠 live 或 persisted target 的 inbox／历史状态。每个当前版本 Team payload 都会经过运行时验证后才进入投影状态。Team runtime 从同步准入到 settlement 全程跟踪 dispatch 与异步 acknowledgement 工作；dispose 会关闭准入，并在移除服务前等待两者。当前 waiter 只在所属 Team event flush 成功后被唤醒。 对于 live target，quiet `send_message` 会立即注入、flush 并确认，但不会唤醒它；inactive target 会保持 queued，直到其他事件 materialize 该 teammate。waking `followup_task` 成为 target 的下一个 FIFO turn，并可冷恢复。即使即时投递被推迟，成功也表示消息已经持久化。该机制提供进程内重试与 target Session 去重，不宣称跨进程 exactly-once。 @@ -62,7 +62,7 @@ Worktree isolation 不是 harness runtime 行为。deployment 或 prompt 可以 ## Testing -Package test 以逐文件 100% coverage 覆盖身份、名字与权限检查、provider 选择、预留 id 持久化冲突、child-before-Lead flush 顺序、持久 provisioning 失败与 pending-inbox JSONL／SQLite 对账、target-local 并发顺序、pending／history 去重、mailbox 限额、flush 后 notification、取消在途创建与 dispatch 的有界 dispose、failed member cleanup、task CAS 与 DAG 校验、write-scope warning、wait cancel／timeout、保留 inbox 的 interrupt、普通 fork 隔离、旧 control shadowing、声明 schema 的紧凑结果渲染与 scoped registration HMR。一条 keyless headless Loader 快照会组合真实 Team 插件，并记录 teammate 创建、peer mail、依赖任务、等待与 Lead 汇总。 +Package test 以逐文件 100% coverage 覆盖身份、名字与权限检查、provider 选择、预留 id 持久化冲突、child-before-Lead flush 顺序、持久 provisioning 失败与 pending-inbox JSONL／SQLite 对账、target-local 并发顺序、pending／history 去重、mailbox 限额、flush 后 notification、取消在途创建与 dispatch 的有界 dispose、failed member cleanup、task CAS 与 DAG 校验、write-scope warning、wait cancel／timeout、保留 inbox 的 interrupt、Team projection checkpoint 状态、普通 fork 隔离、旧 control shadowing、声明 schema 的紧凑结果渲染与 scoped registration HMR。一条 keyless headless Loader 快照会组合真实 Team 插件，并记录 teammate 创建、peer mail、依赖任务、等待与 Lead 汇总。 ## Consequences diff --git a/packages/experimental/team/README.i18n.yaml b/packages/experimental/team/README.i18n.yaml index 679ab3418b..629ae1841b 100644 --- a/packages/experimental/team/README.i18n.yaml +++ b/packages/experimental/team/README.i18n.yaml @@ -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/experimental/team/README.md -README.md: f676f5b2245fd15559c9e5cd5f969ceece6625c8 -README.zh.md: e85dc59dcd349f69d23408038b4ecead01fbeca6 +README.md: 2649752c7cabec1e71fc6e2570d4ec36670e67ba +README.zh.md: de1acb2ffda938387637cb94e4c776e49c9f221e diff --git a/packages/experimental/team/README.md b/packages/experimental/team/README.md index f676f5b224..2649752c7c 100644 --- a/packages/experimental/team/README.md +++ b/packages/experimental/team/README.md @@ -19,11 +19,11 @@ Implicit-root Agent Teams domain. `ctx.teams` owns a flat Lead/teammate roster, Every limit must be a positive safe integer. `maxMembers` counts every name ever provisioned, including failed members, because names are never reusable. `maxTasks` counts non-deleted tasks. The mailbox limit is per target; the byte limit covers the complete framed delivery, including its stable id and sender name. `disposalTimeoutMs` bounds admitted creation, mailbox dispatch, and Team-owned Activation settlement so plugin reload and process shutdown fail visibly instead of waiting forever. -The service requires Agent, Session, Session persistence, and continuable-subagent services. A composition without durable Session storage does not activate it. +The service requires Agent, Session, Session persistence, Session projection, and continuable-subagent services. A composition without durable Session storage or the projection registry does not activate it. ## Team identity and roster -Every ordinary runtime root is the implicit Lead of a Team whose `TeamId` equals its `SessionId`; creating a Team is therefore state-free until the first member, message, or task record. A teammate is a named, continuable direct child recorded in that root's Session. Names are lowercase kebab-case, at most 64 characters, and immutable for the Team lifetime. Session ids remain the persistence and authorization identities. +Every ordinary runtime root is the implicit Lead of a Team whose `TeamId` equals its `SessionId`; creating a Team is therefore state-free until the first member, message, or task record. The host-only Team projection folds committed records eagerly into checkpoint-safe JSON grouped by `TeamId`; Team reads select the current Lead's group, so an ordinary fork cannot adopt inherited parent state. A teammate is a named, continuable direct child recorded in that root's Session. Names are lowercase kebab-case, at most 64 characters, and immutable for the Team lifetime. Session ids remain the persistence and authorization identities. `spawnTeammate()` first appends and flushes a provisioning member, then asks the configured spawn or fork provider to create the reserved child id. A provider failure appends a durable failed member. Successful inbox admission is flushed in the child Session before the active edge commits. On root recovery, a provisioning record becomes active only when the independently persisted child has matching direct-parent and continuable descriptors plus its initial user message, either still pending in the durable inbox or already recorded in history; otherwise it becomes failed. If recovery wins a same-process provisioning race, the creator accepts the matching terminal state or reports `TEAM_PROVISIONING_CONFLICT` and drains a child that recovery already marked failed. Disposal closes admission, aborts and awaits admitted creation and mailbox-dispatch transactions, then asks the continuation owner to release the roster's exact live direct children and their descendants. Non-Team continuable children of the Lead remain untouched. Cleanup failures make disposal fail visibly. This closes crashes and reload races between root provisioning and its terminal member edge without reusing a name or retaining an orphan Activation. @@ -49,8 +49,6 @@ Dependencies must name current non-deleted tasks and form a complete DAG with no `waitForChange()` waits for one roster, task, mailbox, or live-status edge that occurs after registration, for 10 seconds through one hour; it reports only whether the wait timed out and does not replay a change that already happened. Runtime disposal releases current waits and makes later waits return immediately without a timeout. Callers re-read authoritative state after wakeup or timeout. Cancellation preserves an Error reason or reports a non-Error reason through `TEAM_WAIT_ABORTED` with structural inspection instead of object coercion. `interrupt()` is Lead-only and delegates to the continuable-subagent interrupt path, which cancels only a live teammate's current turn with `keepInbox`; it neither releases task ownership nor deletes durable mail. -The separate `./invariant` companion replays each candidate Team event against its committed Session prefix. Replay validates every current-version Team payload before it enters folded state, then rejects invalid member transitions, reused names, out-of-range numeric task ids, discontinuous task revisions, invalid task dependencies, duplicate queue/ack records, and acknowledgements with the wrong target before append. Session event `seq` and `time` own ordering and timing instead of duplicated snapshot timestamps. - ## Model Experience ### Peer messages diff --git a/packages/experimental/team/README.zh.md b/packages/experimental/team/README.zh.md index e85dc59dcd..de1acb2ffd 100644 --- a/packages/experimental/team/README.zh.md +++ b/packages/experimental/team/README.zh.md @@ -19,11 +19,11 @@ 每个限制都必须是正的安全整数。`maxMembers` 统计所有曾 provision 的名字，包括失败成员，因为名字永不复用。`maxTasks` 统计未删除任务。mailbox 限额按目标成员计算；字节限制覆盖完整的投递帧，包括稳定 id 与发送者名称。`disposalTimeoutMs` 限制已获准创建、mailbox dispatch 与 Team 自有 Activation 的 settlement 时长，使插件 reload 与进程 shutdown 在异常时明确失败，而不是无限等待。 -该服务要求 Agent、Session、Session persistence 与 continuable-subagent 服务。没有持久 Session 存储的组合不会激活它。 +该服务要求 Agent、Session、Session persistence、Session projection 与 continuable-subagent 服务。缺少持久 Session 存储或 projection registry 的组合不会激活它。 ## Team 身份与 roster -每个普通运行时 Root 都是一个隐式 Team 的 Lead，其 `TeamId` 等于 `SessionId`；因此，在写入第一条成员、消息或任务记录前，创建 Team 不需要额外状态。teammate 是记录在 Root Session 中的具名 continuable 直接 child。名字采用小写 kebab-case，最长 64 个字符，在 Team 生命周期内不可变。Session id 始终是持久化与授权身份。 +每个普通运行时 Root 都是一个隐式 Team 的 Lead，其 `TeamId` 等于 `SessionId`；因此，在写入第一条成员、消息或任务记录前，创建 Team 不需要额外状态。host-only Team projection 会把已提交记录即时折叠成按 `TeamId` 分组且可安全 checkpoint 的 JSON；Team 读取只选择当前 Lead 的分组，因此普通 fork 不会采用继承的 parent 状态。teammate 是记录在 Root Session 中的具名 continuable 直接 child。名字采用小写 kebab-case，最长 64 个字符，在 Team 生命周期内不可变。Session id 始终是持久化与授权身份。 `spawnTeammate()` 先追加并 flush provisioning member，再要求配置的 spawn 或 fork provider 使用预留 child id 创建成员。provider 失败会追加持久 failed member。初始 inbox 消息获准后，先 flush child Session，再提交 active 边。Root 恢复时，只有独立持久 child 的直接 parent 与 continuable descriptor 匹配，并且其初始用户消息仍在持久 inbox 中或已经记录进历史，provisioning 才转为 active；否则转为 failed。如果 recovery 在同进程 provisioning 竞争中先完成，creator 会接受匹配终态，或报告 `TEAM_PROVISIONING_CONFLICT` 并 drain 已被 recovery 标为 failed 的 child。dispose 会关闭准入，中止并等待已获准的创建与 mailbox dispatch 事务，再让 continuation owner 释放 roster 中确切的 live direct child 及其后代；Lead 的非 Team continuable child 不受影响。cleanup 失败会让 dispose 明确失败。该对账覆盖 Root provisioning 与终态成员边之间的崩溃和 reload 窗口，同时不复用名字或遗留孤儿 Activation。 @@ -49,8 +49,6 @@ roster 同时报告持久 provisioning／failed phase 与实时 `running`／`idl `waitForChange()` 可以等待注册后发生的下一条 roster、task、mailbox 或实时 status 边，时长范围为 10 秒到 1 小时；它只报告等待是否超时，也不会回放调用前已经发生的变化。运行时 dispose 会释放当前等待，并使后续等待不经超时立即返回。调用方需要在唤醒或超时后重新读取权威状态。取消会保留 Error reason；非 Error reason 则通过 `TEAM_WAIT_ABORTED` 以结构化检查结果报告，不再强制转成 object 字符串。`interrupt()` 仅限 Lead，并委托 continuable-subagent 的 interrupt 路径以 `keepInbox` 只取消 live teammate 的当前 turn；它既不释放任务 owner，也不删除持久 mail。 -独立的 `./invariant` 配套模块会把每条候选 Team event 对照已提交 Session 前缀回放。回放会先验证每个当前版本 Team payload，再将其纳入折叠状态；随后会在 append 前拒绝非法 member 转换、名字复用、超出范围的数字 task id、不连续任务 revision、非法任务依赖、重复 queue／ack，以及 target 不匹配的 acknowledgement。顺序与时间由 Session event 的 `seq` 和 `time` 负责，不在 snapshot 中重复保存。 - ## 模型体验 ### Peer 消息 diff --git a/packages/experimental/team/package.json b/packages/experimental/team/package.json index f68d1da12f..86efb33291 100644 --- a/packages/experimental/team/package.json +++ b/packages/experimental/team/package.json @@ -40,6 +40,7 @@ "@deepseek-ai/dsh-llm": "workspace:^", "@deepseek-ai/dsh-session": "workspace:^", "@deepseek-ai/dsh-session-persistence": "workspace:^", + "@deepseek-ai/dsh-session-projection": "workspace:^", "@deepseek-ai/dsh-subagent": "workspace:^", "@deepseek-ai/cordis": "workspace:^" }, diff --git a/packages/experimental/team/src/index.ts b/packages/experimental/team/src/index.ts index 9b80d4d8dc..a3e31efc4c 100644 --- a/packages/experimental/team/src/index.ts +++ b/packages/experimental/team/src/index.ts @@ -9,6 +9,7 @@ import { errorMessage, TeamError } from './error.ts' import { TeamJournal } from './journal.ts' import { TeamRuntimeLifecycle } from './lifecycle.ts' import { TeamMailbox } from './mailbox.ts' +import { teamProjectionDefinition } from './projection.ts' import { TeamRoster } from './roster.ts' import type { TeamMembership } from './roster.ts' import { TeamTaskBoard } from './task-board.ts' @@ -30,7 +31,6 @@ export type * from './types.ts' export type { TeamMembership } from './roster.ts' export { TeamId, TeamMessageId, TeamTaskId } from './types.ts' export { TeamError } from './error.ts' -export { foldTeam } from './fold.ts' declare module '@deepseek-ai/cordis' { interface Context { @@ -54,7 +54,7 @@ function positiveLimit(name: string, value: number): number { /** Agent Teams service backed by the exact live Lead Session log. */ export class TeamService extends Service { - static inject = ['agents', 'sessions', 'sessionPersistence', 'subagents'] + static inject = ['agents', 'sessions', 'sessionPersistence', 'sessionProjections', 'subagents'] static Config: z = z.object({ maxMembers: z.number().step(1).min(1).default(DEFAULT_MAX_MEMBERS), @@ -110,7 +110,10 @@ export class TeamService extends Service { const membership = this.roster.tryMembership(agent) if (membership !== undefined) this.activity.notify(membership.id) }) - ctx.effect(() => () => this.disposeRuntime(), 'teams.runtimeLifecycle()') + ctx.effect(function* (this: TeamService) { + yield ctx.sessionProjections.register(teamProjectionDefinition) + yield () => this.disposeRuntime() + }.bind(this), 'teams.runtimeLifecycle()') for (const agent of ctx.agents.list()) this.scheduleRecovery(agent) } diff --git a/packages/experimental/team/src/invariant.ts b/packages/experimental/team/src/invariant.ts index 52e28491fe..ddd18ad208 100644 --- a/packages/experimental/team/src/invariant.ts +++ b/packages/experimental/team/src/invariant.ts @@ -1,9 +1,7 @@ -/** Package-owned relational checks for Agent Teams durable records. */ +/** Agent Teams runtime invariant companion. */ import type { Context } from '@deepseek-ai/cordis' -import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants' -import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' -import { applyTeamEvent, foldTeam, isTeamEvent } from './fold.ts' +import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants' const PACKAGE_NAME = '@deepseek-ai/dsh-team' @@ -12,22 +10,8 @@ export const name = 'team-invariant' /** Invariant registry required by the companion. */ export const inject = ['invariants'] -/** Validate candidate Team events against the committed prefix before append. */ -const install: InvariantInstaller = Object.assign((ctx: Context, fail: InvariantFailure) => { - ctx.on('internal/dispatch', (_mode, eventName, args) => { - if (eventName !== 'session/event') return - const [session, event] = args as [Session, SessionEvent] - if (!isTeamEvent(event)) return - try { - const state = foldTeam(session.id, session.events) - applyTeamEvent(state, event) - } catch (error: unknown) { - /* v8 ignore next -- the strict Team fold throws Error instances. */ - const message = error instanceof Error ? error.message : String(error) - fail(`session event ${event.seq} violates the Agent Teams stream: ${message}`) - } - }, { global: true }) -}, { inject: ['sessions'] }) +/** No runtime invariant: the Team projection owns event decoding and relational state transitions. */ +const install: InvariantInstaller = () => {} /** Register the package invariant companion. */ export const apply = (ctx: Context): Promise<() => void> => diff --git a/packages/experimental/team/src/journal.ts b/packages/experimental/team/src/journal.ts index f97e7776f9..391966bcbc 100644 --- a/packages/experimental/team/src/journal.ts +++ b/packages/experimental/team/src/journal.ts @@ -3,8 +3,9 @@ import type { Agent } from '@deepseek-ai/dsh-agent' import type { Context } from '@deepseek-ai/cordis' import type { SessionEventMap, SessionId } from '@deepseek-ai/dsh-session' -import { foldTeam } from './fold.ts' -import type { TeamEventType, TeamFoldState } from './fold.ts' +import { emptyTeamState } from './projection.ts' +import type { TeamEventType, TeamState } from './projection.ts' +import { TeamId } from './types.ts' type AppendTeamEvent = (type: T, data: SessionEventMap[T]) => void type MutableTeamEventType = 'team/member' | 'team/task' | 'team/message/queued' | 'team/message/delivered' @@ -23,12 +24,16 @@ export class TeamJournal { ) {} /** - * Fold authoritative Team state for one exact live Lead. + * Read authoritative Team state for one exact live Lead. * @param root - exact live Team Lead. - * @returns current replay state selected by the Lead Team id. + * @returns current projected state selected by the Lead Team id. */ - state(root: Agent): TeamFoldState { - return foldTeam(root.id, root.session.events) + state(root: Agent): TeamState { + const projection = this.ctx.sessionProjections.stateOf(root.session, 'team') + if (projection === undefined) throw new Error('Agent Teams projection is not registered') + const selected = projection.teams.find(team => team.id === TeamId(root.id)) + if (selected?.failure !== undefined) throw new Error(selected.failure) + return selected ?? emptyTeamState(root.id) } /** diff --git a/packages/experimental/team/src/mailbox.ts b/packages/experimental/team/src/mailbox.ts index b234cef408..436e494fb1 100644 --- a/packages/experimental/team/src/mailbox.ts +++ b/packages/experimental/team/src/mailbox.ts @@ -86,8 +86,8 @@ export class TeamMailbox { const membership = this.roster.tryMembership(agent) if (membership === undefined) return const state = this.journal.state(membership.root) - const messages = [...state.messages.values()].filter(message => - !state.delivered.has(message.id) + const messages = state.messages.filter(message => + !state.delivered.includes(message.id) && (membership.role === 'lead' || message.targetId === agent.id)) for (const message of messages) { signal.throwIfAborted() @@ -119,8 +119,8 @@ export class TeamMailbox { const state = this.journal.state(root) const target = resolveActiveMember(root, state, request.target) if (target.id === caller.id) throw new TeamError('a Team member cannot message itself', 'TEAM_SELF_MESSAGE') - const pendingForTarget = [...state.messages.values()].filter(candidate => - candidate.targetId === target.id && !state.delivered.has(candidate.id)).length + const pendingForTarget = state.messages.filter(candidate => + candidate.targetId === target.id && !state.delivered.includes(candidate.id)).length if (pendingForTarget >= this.maxPendingMessagesPerMember) { throw new TeamError( `teammate "${target.name}" has ${pendingForTarget} pending messages`, @@ -272,7 +272,7 @@ export class TeamMailbox { /** Whether `left` was durably queued before `right` in one Lead log. */ private messagePrecedes(root: Agent, left: TeamMessageId, right: TeamMessageId): boolean { - const ids = [...this.journal.state(root).messages.keys()] + const ids = this.journal.state(root).messages.map(message => message.id) return ids.indexOf(left) < ids.indexOf(right) } @@ -292,8 +292,8 @@ export class TeamMailbox { private async markDelivered(root: Agent, messageId: TeamMessageId, targetId: SessionId): Promise { await this.journal.transact(root.id, async () => { const state = this.journal.state(root) - if (state.delivered.has(messageId)) return - const queued = state.messages.get(messageId) + if (state.delivered.includes(messageId)) return + const queued = state.messages.find(message => message.id === messageId) if (queued === undefined || queued.targetId !== targetId) return await this.journal.appendAndFlush(root, 'team/message/delivered', { version: 1, diff --git a/packages/experimental/team/src/fold.ts b/packages/experimental/team/src/projection.ts similarity index 65% rename from packages/experimental/team/src/fold.ts rename to packages/experimental/team/src/projection.ts index cf439a5322..1ab43a0b53 100644 --- a/packages/experimental/team/src/fold.ts +++ b/packages/experimental/team/src/projection.ts @@ -1,15 +1,15 @@ -/** Strict replay fold for Agent Teams log-only events. */ +/** Host-only Team state projected incrementally from committed Session events. */ import { z } from 'zod' import type { ContentBlock } from '@deepseek-ai/dsh-llm' import { SessionId } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionEventMap } from '@deepseek-ai/dsh-session' +import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection' import type { TeamId, TeamMemberSnapshot, TeamMessageId, TeamMessageSnapshot, - TeamTaskId, TeamTaskSnapshot, } from './types.ts' import { @@ -124,34 +124,62 @@ const teamMessageDeliveredEventSchema = z.object({ targetId: sessionIdSchema, }).strict() as z.ZodType -/** Mutable internal replay state. */ -export interface TeamFoldState { +/** Current Team state selected by durable Team identity. */ +export interface TeamState { readonly id: TeamId - readonly members: Map - readonly memberIdsByName: Map - readonly tasks: Map - readonly messages: Map - readonly delivered: Set + readonly members: TeamMemberSnapshot[] + readonly tasks: TeamTaskSnapshot[] + readonly messages: TeamMessageSnapshot[] + readonly delivered: TeamMessageId[] nextTaskNumber: number } /** - * Construct an empty Team fold for one root Session. - * @param rootId - Session whose TeamId selects applicable records. - * @returns mutable empty replay state. + * Construct empty state for one Team identity. + * @param rootId - root Session identity. + * @returns mutable empty Team state. */ -export function emptyTeamFoldState(rootId: SessionId): TeamFoldState { +export function emptyTeamState(rootId: SessionId): TeamProjectionEntry { return { id: toTeamId(rootId), - members: new Map(), - memberIdsByName: new Map(), - tasks: new Map(), - messages: new Map(), - delivered: new Set(), + members: [], + tasks: [], + messages: [], + delivered: [], nextTaskNumber: 1, } } +/** One checkpoint-safe Team state retained by its durable identity. */ +export interface TeamProjectionEntry extends TeamState { + failure?: string +} + +/** Plain-JSON Team states grouped by identity for fork isolation. */ +export interface TeamProjectionState { + readonly teams: TeamProjectionEntry[] +} + +declare module '@deepseek-ai/dsh-session-projection/types' { + interface SessionProjectionStateMap { + team: TeamProjectionState + } +} + +const teamProjectionEntrySchema = z.object({ + id: teamIdSchema, + members: z.array(teamMemberSnapshotSchema), + tasks: z.array(teamTaskSnapshotSchema), + messages: z.array(teamMessageSnapshotSchema), + delivered: z.array(teamMessageIdSchema), + nextTaskNumber: positiveSafeInteger, + failure: z.string().optional(), +}).strict() as z.ZodType + +const teamProjectionStateSchema = z.object({ + teams: z.array(teamProjectionEntrySchema), +}).strict() as z.ZodType + /** Whether one event belongs to the Team domain. */ export type TeamEventType = | 'team/member' @@ -160,14 +188,14 @@ export type TeamEventType = | 'team/message/delivered' /** One event owned by the Team domain. */ -export type TeamSessionEvent = SessionEvent +type TeamSessionEvent = SessionEvent /** * Test whether a Session event belongs to the Team domain. * @param event - candidate Session event. * @returns whether the event has a Team-owned type. */ -export function isTeamEvent(event: SessionEvent): event is TeamSessionEvent { +function isTeamEvent(event: SessionEvent): event is TeamSessionEvent { return event.type === 'team/member' || event.type === 'team/task' || event.type === 'team/message/queued' @@ -200,32 +228,39 @@ function parseCurrentTeamEvent(event: TeamSessionEvent): TeamSessionEvent { } } -/** - * Apply one event, ignoring Team records inherited by a different root fork. - * @param state - mutable Team replay state. - * @param event - next contiguous Session event. - */ -export function applyTeamEvent(state: TeamFoldState, event: SessionEvent): void { +function applyProjectionEvent(state: TeamProjectionState, event: SessionEvent): void { if (!isTeamEvent(event)) return const selector = parsePersisted(event.type, teamEventSelectorSchema, event.data) - if (selector.version !== 1) { - if (selector.teamId !== state.id) return - throw new Error(`unsupported Agent Teams event version ${String(selector.version)}`) + const current = selector.version === 1 ? parseCurrentTeamEvent(event) : undefined + let team = state.teams.find(candidate => candidate.id === selector.teamId) + if (team === undefined) { + team = emptyTeamState(SessionId(selector.teamId)) + state.teams.push(team) } - const decoded = parseCurrentTeamEvent(event) - if (decoded.data.teamId !== state.id) return + if (team.failure !== undefined) return + try { + if (current === undefined) { + throw new Error(`unsupported Agent Teams event version ${String(selector.version)}`) + } + applyCurrentTeamEvent(team, current) + } catch (error: unknown) { + /* v8 ignore next -- the owned Team transition throws Error instances. */ + team.failure = error instanceof Error ? error.message : String(error) + } +} - switch (decoded.type) { +function applyCurrentTeamEvent(state: TeamState, event: TeamSessionEvent): void { + switch (event.type) { case 'team/member': { - const member = decoded.data.member - const prior = state.members.get(member.id) - const named = state.memberIdsByName.get(member.name) - if (named !== undefined && named !== member.id) { + const member = event.data.member + const index = state.members.findIndex(candidate => candidate.id === member.id) + const prior = state.members[index] + const named = state.members.find(candidate => candidate.name === member.name) + if (named !== undefined && named.id !== member.id) { throw new Error(`teammate name "${member.name}" is reused by another member`) } if (prior === undefined) { if (member.phase !== 'provisioning') throw new Error(`teammate "${member.name}" must begin provisioning`) - state.memberIdsByName.set(member.name, member.id) } else { if (prior.name !== member.name || prior.provider !== member.provider || prior.context !== member.context) { throw new Error(`teammate "${member.id}" changed immutable identity fields`) @@ -234,12 +269,14 @@ export function applyTeamEvent(state: TeamFoldState, event: SessionEvent): void throw new Error(`teammate "${member.name}" has an invalid ${prior.phase} -> ${member.phase} transition`) } } - state.members.set(member.id, member) + if (index < 0) state.members.push(member) + else state.members[index] = member break } case 'team/task': { - const task = decoded.data.task - const prior = state.tasks.get(task.id) + const task = event.data.task + const index = state.tasks.findIndex(candidate => candidate.id === task.id) + const prior = state.tasks[index] if (prior === undefined && task.revision !== 1) { throw new Error(`team task "${task.id}" must begin at revision 1`) } @@ -255,21 +292,24 @@ export function applyTeamEvent(state: TeamFoldState, event: SessionEvent): void number === Number.MAX_SAFE_INTEGER ? number : number + 1, ) } - state.tasks.set(task.id, task) + if (index < 0) state.tasks.push(task) + else state.tasks[index] = task break } case 'team/message/queued': { - const message = decoded.data.message - if (state.messages.has(message.id)) throw new Error(`team message "${message.id}" was queued twice`) - state.messages.set(message.id, message) + const message = event.data.message + if (state.messages.some(candidate => candidate.id === message.id)) { + throw new Error(`team message "${message.id}" was queued twice`) + } + state.messages.push(message) break } case 'team/message/delivered': { - const queued = state.messages.get(decoded.data.messageId) - if (queued === undefined) throw new Error(`team message "${decoded.data.messageId}" was delivered before queueing`) - if (queued.targetId !== decoded.data.targetId) throw new Error(`team message "${decoded.data.messageId}" target changed`) - if (state.delivered.has(decoded.data.messageId)) throw new Error(`team message "${decoded.data.messageId}" was delivered twice`) - state.delivered.add(decoded.data.messageId) + const queued = state.messages.find(message => message.id === event.data.messageId) + if (queued === undefined) throw new Error(`team message "${event.data.messageId}" was delivered before queueing`) + if (queued.targetId !== event.data.targetId) throw new Error(`team message "${event.data.messageId}" target changed`) + if (state.delivered.includes(event.data.messageId)) throw new Error(`team message "${event.data.messageId}" was delivered twice`) + state.delivered.push(event.data.messageId) break } /* v8 ignore next 2 -- TeamEventType is closed and every member is handled above. */ @@ -278,14 +318,14 @@ export function applyTeamEvent(state: TeamFoldState, event: SessionEvent): void } } -/** - * Replay one root Session into its current Team state. - * @param rootId - root Session identity selecting Team-owned records. - * @param events - complete contiguous Session log. - * @returns mutable replay state at the end of the log. - */ -export function foldTeam(rootId: SessionId, events: readonly SessionEvent[]): TeamFoldState { - const state = emptyTeamFoldState(rootId) - for (const event of events) applyTeamEvent(state, event) - return state -} +/** Host-only Team projection grouped by durable Team identity. */ +export const teamProjectionDefinition = { + key: 'team', + stateVersion: 1, + stateSchema: teamProjectionStateSchema, + init: (): TeamProjectionState => ({ teams: [] }), + apply: (state, event) => { + applyProjectionEvent(state, event) + return state + }, +} satisfies ProjectionDefinition<'team', TeamProjectionState> diff --git a/packages/experimental/team/src/roster.ts b/packages/experimental/team/src/roster.ts index 2d8d190fce..907df82daa 100644 --- a/packages/experimental/team/src/roster.ts +++ b/packages/experimental/team/src/roster.ts @@ -8,9 +8,9 @@ import { SessionId } from '@deepseek-ai/dsh-session' import { foldSubagentDescriptor } from '@deepseek-ai/dsh-subagent' import type { ContinuableStart } from '@deepseek-ai/dsh-subagent' import { errorMessage, TeamError } from './error.ts' -import type { TeamFoldState } from './fold.ts' import type { TeamJournal } from './journal.ts' import type { TeamRuntimeLifecycle } from './lifecycle.ts' +import type { TeamState } from './projection.ts' import { messageAccepted } from './session-message.ts' import { TeamId } from './types.ts' import type { @@ -34,19 +34,18 @@ export interface TeamMembership { /** * Resolve one active Team member by model-facing name, including the Lead pseudo-row. * @param root - exact live Team Lead. - * @param state - current Team fold. + * @param state - current Team state. * @param rawName - candidate member name. * @returns resolved durable id and normalized name. */ export function resolveActiveMember( root: Agent, - state: TeamFoldState, + state: TeamState, rawName: string, ): { id: SessionId; name: string } { const name = rawName.trim() if (name === 'lead') return { id: root.id, name } - const id = state.memberIdsByName.get(name) - const member = id === undefined ? undefined : state.members.get(id) + const member = state.members.find(candidate => candidate.name === name) if (member === undefined || member.phase !== 'active') { throw new TeamError(`active teammate "${name}" not found`, 'TEAM_MEMBER_NOT_FOUND') } @@ -95,7 +94,7 @@ export class TeamRoster { if (parentId !== undefined) { const root = this.ctx.agents.get(parentId) if (root !== undefined) { - const member = this.journal.state(root).members.get(agent.id) + const member = this.journal.state(root).members.find(candidate => candidate.id === agent.id) if (member?.phase === 'active' || member?.phase === 'provisioning') { return { root, id: TeamId(root.id), role: 'teammate', name: member.name } } @@ -109,7 +108,7 @@ export class TeamRoster { // A continuation can briefly outlive its parent during child-first teardown. // Do not reinterpret that durable child as a new implicit root Team. A host- // resumed ordinary fork has no descriptor in its own suffix and remains a - // valid new root whose inherited Team records fold out by TeamId. + // valid new root whose inherited Team records stay outside its projected Team state. if (this.subagentDescriptor(agent)) return undefined return { root: agent, id: TeamId(agent.id), role: 'lead', name: 'lead' } } catch { @@ -136,7 +135,7 @@ export class TeamRoster { ...root.options.model === undefined ? {} : { model: root.options.model }, diagnostics: [], }] - for (const member of state.members.values()) { + for (const member of state.members) { const live = this.ctx.agents.get(member.id) const model = live?.options.model ?? root.options.model result.push({ @@ -223,7 +222,8 @@ export class TeamRoster { const rootId = agent.session.header.parentSession if (rootId === undefined) continue const root = this.ctx.agents.get(rootId) - if (root === undefined || !this.journal.state(root).members.has(agent.id)) continue + if (root === undefined + || !this.journal.state(root).members.some(member => member.id === agent.id)) continue const children = teams.get(root) ?? [] children.push(agent.id) teams.set(root, children) @@ -266,10 +266,10 @@ export class TeamRoster { await this.journal.transact(root.id, async () => { const state = this.journal.state(root) - if (state.memberIdsByName.has(name)) { + if (state.members.some(member => member.name === name)) { throw new TeamError(`teammate name "${name}" was already used in this Team`, 'TEAM_MEMBER_NAME_TAKEN') } - if (state.members.size >= this.maxMembers) { + if (state.members.length >= this.maxMembers) { throw new TeamError(`Team member limit ${this.maxMembers} reached`, 'TEAM_MEMBER_LIMIT') } await this.journal.appendAndFlush(root, 'team/member', { version: 1, teamId: TeamId(root.id), member }) @@ -387,7 +387,7 @@ export class TeamRoster { /** Settle provisioning-only members from their independently durable child Sessions. */ private async reconcileProvisioning(root: Agent, signal: AbortSignal): Promise { - const provisioning = [...this.journal.state(root).members.values()].filter(member => member.phase === 'provisioning') + const provisioning = this.journal.state(root).members.filter(member => member.phase === 'provisioning') for (const member of provisioning) { signal.throwIfAborted() // A live child means creation is still completing in this process. Its @@ -414,7 +414,7 @@ export class TeamRoster { signal.throwIfAborted() await this.journal.transact(root.id, async () => { signal.throwIfAborted() - const current = this.journal.state(root).members.get(member.id) + const current = this.journal.state(root).members.find(candidate => candidate.id === member.id) if (current?.phase !== 'provisioning') return const settled: TeamMemberSnapshot = { ...current, @@ -463,7 +463,7 @@ export class TeamRoster { terminal: TeamMemberSnapshot, ): Promise<'active' | 'failed'> { return this.journal.transact(root.id, async () => { - const current = this.journal.state(root).members.get(terminal.id) + const current = this.journal.state(root).members.find(member => member.id === terminal.id) /* v8 ignore next 3 -- the append-only provisioning event is committed by this operation before settlement. */ if (current === undefined) { throw new TeamError(`provisioned teammate "${terminal.id}" disappeared`, 'TEAM_PROVISIONING_CONFLICT') diff --git a/packages/experimental/team/src/task-board.ts b/packages/experimental/team/src/task-board.ts index f987dd9217..5dd235f940 100644 --- a/packages/experimental/team/src/task-board.ts +++ b/packages/experimental/team/src/task-board.ts @@ -3,8 +3,8 @@ import type { Agent } from '@deepseek-ai/dsh-agent' import type { TeamMembership } from './roster.ts' import { TeamError } from './error.ts' -import type { TeamFoldState } from './fold.ts' import type { TeamJournal } from './journal.ts' +import type { TeamState } from './projection.ts' import { resolveActiveMember } from './roster.ts' import { assertTaskGraphCandidate, TeamTaskGraphError } from './task-graph.ts' import type { TeamTaskGraphViolation } from './task-graph.ts' @@ -49,12 +49,12 @@ export class TeamTaskBoard { const { root } = membership return this.journal.transact(root.id, async () => { const state = this.journal.state(root) - const active = [...state.tasks.values()].filter(task => task.status !== 'deleted').length + const active = state.tasks.filter(task => task.status !== 'deleted').length if (active >= this.maxTasks) { throw new TeamError(`Team task limit ${this.maxTasks} reached`, 'TEAM_TASK_LIMIT') } const id = TeamTaskId(`task-${state.nextTaskNumber}`) - if (state.tasks.has(id)) { + if (state.tasks.some(task => task.id === id)) { throw new TeamError('Team task id space exhausted', 'TEAM_TASK_LIMIT') } const task: TeamTaskSnapshot = { @@ -81,7 +81,7 @@ export class TeamTaskBoard { get(membership: TeamMembership, id: TeamTaskId): TeamTaskView { const { root } = membership const state = this.journal.state(root) - const task = state.tasks.get(id) + const task = state.tasks.find(candidate => candidate.id === id) if (task === undefined) throw new TeamError(`team task "${id}" not found`, 'TEAM_TASK_NOT_FOUND') return this.taskView(root, state, task) } @@ -94,7 +94,7 @@ export class TeamTaskBoard { list(membership: TeamMembership): TeamTaskView[] { const { root } = membership const state = this.journal.state(root) - return [...state.tasks.values()] + return state.tasks .filter(task => task.status !== 'deleted') .map(task => this.taskView(root, state, task)) } @@ -114,7 +114,7 @@ export class TeamTaskBoard { const root = membership.root return this.journal.transact(root.id, async () => { const state = this.journal.state(root) - const current = state.tasks.get(request.taskId) + const current = state.tasks.find(task => task.id === request.taskId) if (current === undefined) throw new TeamError(`team task "${request.taskId}" not found`, 'TEAM_TASK_NOT_FOUND') if (current.revision !== request.expectedRevision) { throw new TeamError( @@ -192,7 +192,7 @@ export class TeamTaskBoard { } case 'delete': { authorizeOwner() - const dependent = [...state.tasks.values()].find(task => + const dependent = state.tasks.find(task => task.status !== 'deleted' && task.id !== current.id && task.blockedBy.includes(current.id)) if (dependent !== undefined) { throw new TeamError(`team task "${current.id}" still blocks "${dependent.id}"`, 'TEAM_TASK_HAS_DEPENDENTS') @@ -217,7 +217,7 @@ export class TeamTaskBoard { /** Validate and de-duplicate dependency ids against the current task graph. */ private dependencies( values: readonly TeamTaskId[], - state: TeamFoldState, + state: TeamState, self?: TeamTaskId, ): TeamTaskId[] { const seen = new Set() @@ -225,7 +225,7 @@ export class TeamTaskBoard { for (const id of values) { if (id === self) throw new TeamError('a team task cannot block itself', 'TEAM_TASK_DEPENDENCY_CYCLE') if (seen.has(id)) throw new TeamError(`duplicate blocker "${id}"`, 'TEAM_INVALID_ARGUMENT') - const task = state.tasks.get(id) + const task = state.tasks.find(candidate => candidate.id === id) if (task === undefined || task.status === 'deleted') { throw new TeamError(`blocker task "${id}" not found`, 'TEAM_TASK_NOT_FOUND') } @@ -241,7 +241,7 @@ export class TeamTaskBoard { } /** Map shared task-graph validation onto stable command error codes. */ - private assertTaskGraph(state: TeamFoldState, candidate: TeamTaskSnapshot): void { + private assertTaskGraph(state: TeamState, candidate: TeamTaskSnapshot): void { try { assertTaskGraphCandidate(state.tasks, candidate) } catch (error: unknown) { @@ -252,8 +252,8 @@ export class TeamTaskBoard { } /** Whether all current blockers completed. */ - private taskReady(state: TeamFoldState, task: TeamTaskSnapshot): boolean { - return task.blockedBy.every(id => state.tasks.get(id)?.status === 'completed') + private taskReady(state: TeamState, task: TeamTaskSnapshot): boolean { + return task.blockedBy.every(id => state.tasks.find(candidate => candidate.id === id)?.status === 'completed') } /** Remove an optional owner field under exactOptionalPropertyTypes. */ @@ -264,18 +264,18 @@ export class TeamTaskBoard { /** * Build one task view with owner name, readiness, and advisory write overlaps. - * A committing caller may pass its pre-append fold because `task` supplies the + * A committing caller may pass its pre-append state because `task` supplies the * new value explicitly; owner names, blocker readiness, and other task scopes * do not change when that snapshot is appended. */ - private taskView(root: Agent, state: TeamFoldState, task: TeamTaskSnapshot): TeamTaskView { + private taskView(root: Agent, state: TeamState, task: TeamTaskSnapshot): TeamTaskView { const ownerName = task.ownerId === undefined ? undefined : task.ownerId === root.id ? 'lead' - : state.members.get(task.ownerId)?.name + : state.members.find(member => member.id === task.ownerId)?.name const warnings = new Set() - for (const other of state.tasks.values()) { + for (const other of state.tasks) { if (other.id === task.id || other.status !== 'in_progress') continue if (task.writeScopes.some(left => other.writeScopes.some(right => scopesOverlap(left, right)))) { warnings.add(`write scopes overlap with ${other.id}`) diff --git a/packages/experimental/team/src/task-graph.ts b/packages/experimental/team/src/task-graph.ts index 38d9dd144c..bd61b953bc 100644 --- a/packages/experimental/team/src/task-graph.ts +++ b/packages/experimental/team/src/task-graph.ts @@ -24,10 +24,10 @@ export class TeamTaskGraphError extends Error { * @throws {TeamTaskGraphError} when an active dependency is missing, duplicated, self-referential, or cyclic. */ export function assertTaskGraphCandidate( - current: ReadonlyMap, + current: readonly TeamTaskSnapshot[], candidate: TeamTaskSnapshot, ): void { - const tasks = new Map(current) + const tasks = new Map(current.map(task => [task.id, task])) tasks.set(candidate.id, candidate) for (const task of tasks.values()) { diff --git a/packages/experimental/team/tests/invariant.spec.ts b/packages/experimental/team/tests/invariant.spec.ts deleted file mode 100644 index 0bf994ecab..0000000000 --- a/packages/experimental/team/tests/invariant.spec.ts +++ /dev/null @@ -1,70 +0,0 @@ -import { describe, expect, it } from 'vitest' -import { Context } from '@deepseek-ai/cordis' -import InvariantService, { InvariantError } from '@deepseek-ai/dsh-invariants' -import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' -import * as TeamInvariant from '../src/invariant.ts' -import { TeamId, TeamTaskId } from '../src/types.ts' - -async function setup(): Promise { - const ctx = new Context() - await ctx.plugin(SessionStore) - await ctx.plugin(InvariantService, { enabled: true }) - await ctx.plugin(TeamInvariant) - return ctx -} - -describe('Agent Teams stream invariant', () => { - it('accepts provisioning and rejects a terminal member as the first edge', async () => { - const ctx = await setup() - const session = ctx.sessions.create(SessionId('team-invariant')) - const member = { - id: SessionId('team-invariant-child'), - name: 'worker', - description: 'worker responsibility', - provider: 'spawn', - context: 'fresh' as const, - phase: 'provisioning' as const, - } - expect(() => { - session.append('team/member', { version: 1, teamId: TeamId(session.id), member }) - }).not.toThrow() - - const invalid = ctx.sessions.create(SessionId('team-invariant-invalid')) - expect(() => { - invalid.append('team/member', { - version: 1, - teamId: TeamId(invalid.id), - member: { ...member, phase: 'active' }, - }) - }).toThrow(expect.objectContaining>({ - code: 'INVARIANT', - packageName: '@deepseek-ai/dsh-team', - })) - expect(invalid.events).toEqual([]) - }) - - it('rejects an invalid task dependency before publication', async () => { - const ctx = await setup() - const session = ctx.sessions.create(SessionId('team-task-invariant')) - - expect(() => { - session.append('team/task', { - version: 1, - teamId: TeamId(session.id), - task: { - id: TeamTaskId('task-1'), - revision: 1, - subject: 'invalid dependency', - description: 'references a missing blocker', - status: 'pending', - blockedBy: [TeamTaskId('missing')], - writeScopes: [], - }, - }) - }).toThrow(expect.objectContaining>({ - code: 'INVARIANT', - packageName: '@deepseek-ai/dsh-team', - })) - expect(session.events).toEqual([]) - }) -}) diff --git a/packages/experimental/team/tests/persistence.spec.ts b/packages/experimental/team/tests/persistence.spec.ts index cc649d0f5b..f54244c054 100644 --- a/packages/experimental/team/tests/persistence.spec.ts +++ b/packages/experimental/team/tests/persistence.spec.ts @@ -15,7 +15,8 @@ import SqliteSessionPersistence from '@deepseek-ai/dsh-session-persistence-sqlit import SubagentService, { seedDescriptorTurn, snapshotSubagentDescriptor } from '@deepseek-ai/dsh-subagent' import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn-in-process' import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' -import TeamService, { foldTeam, TeamId, TeamMessageId } from '../src/index.ts' +import TeamService, { TeamId, TeamMessageId } from '../src/index.ts' +import { emptyTeamState, teamProjectionDefinition } from '../src/projection.ts' import type { TeamMemberSnapshot, TeamMessageSnapshot, TeamTaskSnapshot } from '../src/index.ts' const SIGNAL = new AbortController().signal @@ -23,17 +24,21 @@ const PERSISTENCE_TEST_TIMEOUT_MS = 15_000 const roots: string[] = [] const contexts = new Set() -/** Detached durable Team read: the service exposes views, so assertions fold the Lead log. */ +/** Detached durable Team read through the same projection definition as the service. */ function durable(agent: Agent): { members: TeamMemberSnapshot[] tasks: TeamTaskSnapshot[] pendingMessages: TeamMessageSnapshot[] } { - const state = foldTeam(agent.id, agent.session.events) + let projected = teamProjectionDefinition.init() + for (const event of agent.session.events) projected = teamProjectionDefinition.apply(projected, event) + const selected = projected.teams.find(team => team.id === TeamId(agent.id)) + if (selected?.failure !== undefined) throw new Error(selected.failure) + const state = selected ?? emptyTeamState(agent.id) return { - members: [...state.members.values()], - tasks: [...state.tasks.values()], - pendingMessages: [...state.messages.values()].filter(message => !state.delivered.has(message.id)), + members: state.members, + tasks: state.tasks, + pendingMessages: state.messages.filter(message => !state.delivered.includes(message.id)), } } diff --git a/packages/experimental/team/tests/fold.spec.ts b/packages/experimental/team/tests/projection-events.spec.ts similarity index 70% rename from packages/experimental/team/tests/fold.spec.ts rename to packages/experimental/team/tests/projection-events.spec.ts index d1ed095a99..b5d050a830 100644 --- a/packages/experimental/team/tests/fold.spec.ts +++ b/packages/experimental/team/tests/projection-events.spec.ts @@ -1,13 +1,8 @@ import { describe, expect, it } from 'vitest' import { SessionId } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionEventMap, SessionEventType } from '@deepseek-ai/dsh-session' -import { - applyTeamEvent, - emptyTeamFoldState, - foldTeam, - isTeamEvent, -} from '../src/fold.ts' -import type { TeamFoldState } from '../src/fold.ts' +import { emptyTeamState, teamProjectionDefinition } from '../src/projection.ts' +import type { TeamProjectionState, TeamState } from '../src/projection.ts' import { TeamId, TeamMessageId, TeamTaskId } from '../src/types.ts' import type { TeamMemberSnapshot, TeamMessageSnapshot, TeamTaskSnapshot } from '../src/types.ts' @@ -19,15 +14,31 @@ function event(type: T, data: SessionEventMap[T], se return { type, data, seq, time: seq } as SessionEvent } -/** Queued-minus-delivered mail, the recovery mailbox the fold is responsible for. */ -function pending(state: TeamFoldState): TeamMessageSnapshot[] { - return [...state.messages.values()].filter(message => !state.delivered.has(message.id)) +function project(events: readonly SessionEvent[]): TeamProjectionState { + let state = teamProjectionDefinition.init() + for (const event of events) state = teamProjectionDefinition.apply(state, event) + return state } -/** Whether one fold reached the end of its log without applying any Team record. */ -function isEmptyFold(state: TeamFoldState): boolean { - return state.members.size === 0 && state.tasks.size === 0 - && state.messages.size === 0 && state.delivered.size === 0 +function teamState(rootId: SessionId, projected: TeamProjectionState): TeamState { + const selected = projected.teams.find(team => team.id === TeamId(rootId)) + if (selected?.failure !== undefined) throw new Error(selected.failure) + return selected ?? emptyTeamState(rootId) +} + +function projectTeam(rootId: SessionId, events: readonly SessionEvent[]): TeamState { + return teamState(rootId, project(events)) +} + +/** Queued-minus-delivered mail retained by the projection. */ +function pending(state: TeamState): TeamMessageSnapshot[] { + return state.messages.filter(message => !state.delivered.includes(message.id)) +} + +/** Whether one Team state contains no projected records. */ +function isEmptyState(state: TeamState): boolean { + return state.members.length === 0 && state.tasks.length === 0 + && state.messages.length === 0 && state.delivered.length === 0 } function member(overrides: Partial = {}): TeamMemberSnapshot { @@ -67,8 +78,8 @@ function message(overrides: Partial = {}): TeamMessageSnaps } } -describe('Agent Teams fold', () => { - it('folds current-team records and ignores inherited records', () => { +describe('Agent Teams projection events', () => { + it('projects current-team records independently from inherited records', () => { const records: SessionEvent[] = [ event('team/member', { version: 1, teamId: TeamId('ancestor'), member: member() }, 0), event('team/member', { version: 1, teamId: TEAM, member: member() }, 1), @@ -80,31 +91,33 @@ describe('Agent Teams fold', () => { event('team/task', { version: 1, teamId: TEAM, task: task({ id: TeamTaskId('task-7') }) }, 3), event('team/message/queued', { version: 1, teamId: TEAM, message: message() }, 4), ] - const state = foldTeam(ROOT, records) + const projected = project(records) + const state = teamState(ROOT, projected) + expect(projected.teams.map(team => team.id)).toEqual([TeamId('ancestor'), TEAM]) expect(state).toMatchObject({ id: TEAM }) - expect(state.members.size).toBe(1) - expect(state.tasks.size).toBe(1) + expect(state.members).toHaveLength(1) + expect(state.tasks).toHaveLength(1) expect(pending(state)).toHaveLength(1) expect(state.nextTaskNumber).toBe(8) - expect(state.members.get(CHILD)?.name).toBe('worker-a') - expect(isTeamEvent(records[0]!)).toBe(true) - expect(isTeamEvent(event('turn/start', { turn: 1 }, 5))).toBe(false) + expect(state.members.find(member => member.id === CHILD)?.name).toBe('worker-a') + expect(teamProjectionDefinition.stateSchema.parse(JSON.parse(JSON.stringify(projected)))) + .toEqual(projected) }) it('enforces teammate identity and lifecycle', () => { const base = event('team/member', { version: 1, teamId: TEAM, member: member() }, 0) - expect(() => foldTeam(ROOT, [event('team/member', { + expect(() => projectTeam(ROOT, [event('team/member', { version: 1, teamId: TEAM, member: member({ phase: 'active' }), }, 0)])).toThrow(/must begin provisioning/) - expect(() => foldTeam(ROOT, [base, event('team/member', { + expect(() => projectTeam(ROOT, [base, event('team/member', { version: 1, teamId: TEAM, member: member({ name: 'renamed', phase: 'active' }), }, 1)])).toThrow(/immutable identity/) - expect(() => foldTeam(ROOT, [base, event('team/member', { + expect(() => projectTeam(ROOT, [base, event('team/member', { version: 1, teamId: TEAM, member: member({ phase: 'active' }), @@ -115,7 +128,7 @@ describe('Agent Teams fold', () => { }, 2)])).toThrow(/invalid active -> failed/) const duplicateName = member({ id: SessionId('child-b') }) - expect(() => foldTeam(ROOT, [base, event('team/member', { + expect(() => projectTeam(ROOT, [base, event('team/member', { version: 1, teamId: TEAM, member: duplicateName, @@ -124,12 +137,12 @@ describe('Agent Teams fold', () => { it('enforces task revision continuity', () => { const first = event('team/task', { version: 1, teamId: TEAM, task: task() }, 0) - expect(() => foldTeam(ROOT, [event('team/task', { + expect(() => projectTeam(ROOT, [event('team/task', { version: 1, teamId: TEAM, task: task({ revision: 2 }), }, 0)])).toThrow(/begin at revision 1/) - expect(() => foldTeam(ROOT, [first, event('team/task', { + expect(() => projectTeam(ROOT, [first, event('team/task', { version: 1, teamId: TEAM, task: task({ revision: 3 }), @@ -189,12 +202,12 @@ describe('Agent Teams fold', () => { ] for (const { records, message: expected } of invalid) { - expect(() => foldTeam(ROOT, records)).toThrow(expected) + expect(() => projectTeam(ROOT, records)).toThrow(expected) } }) it('leaves numeric allocation unchanged for a branded nonstandard task id', () => { - const state = foldTeam(ROOT, [event('team/task', { + const state = projectTeam(ROOT, [event('team/task', { version: 1, teamId: TEAM, task: task({ id: TeamTaskId('external-task') }), @@ -203,7 +216,7 @@ describe('Agent Teams fold', () => { }) it('rejects a persisted numeric task id outside the safe integer range', () => { - expect(() => foldTeam(ROOT, [event('team/task', { + expect(() => projectTeam(ROOT, [event('team/task', { version: 1, teamId: TEAM, task: task({ id: TeamTaskId('task-9007199254740992') }), @@ -218,17 +231,17 @@ describe('Agent Teams fold', () => { messageId: TeamMessageId('message-1'), targetId: CHILD, }, 1) - expect(pending(foldTeam(ROOT, [queued, delivered]))).toEqual([]) - expect(() => foldTeam(ROOT, [queued, queued])).toThrow(/queued twice/) - expect(() => foldTeam(ROOT, [delivered])).toThrow(/delivered before queueing/) - expect(() => foldTeam(ROOT, [queued, event('team/message/delivered', { + expect(pending(projectTeam(ROOT, [queued, delivered]))).toEqual([]) + expect(() => projectTeam(ROOT, [queued, queued])).toThrow(/queued twice/) + expect(() => projectTeam(ROOT, [delivered])).toThrow(/delivered before queueing/) + expect(() => projectTeam(ROOT, [queued, event('team/message/delivered', { ...delivered.data, targetId: SessionId('other'), }, 1)])).toThrow(/target changed/) - expect(() => foldTeam(ROOT, [queued, delivered, { ...delivered, seq: 2 }])).toThrow(/delivered twice/) + expect(() => projectTeam(ROOT, [queued, delivered, { ...delivered, seq: 2 }])).toThrow(/delivered twice/) }) - it('validates every current-version persisted payload before folding it', () => { + it('validates every current-version persisted payload before projecting it', () => { const malformed = [ { ...event('team/member', { version: 1, teamId: TEAM, member: member() }, 0), @@ -271,14 +284,14 @@ describe('Agent Teams fold', () => { ] as unknown as SessionEvent[] for (const candidate of malformed) { - expect(() => foldTeam(ROOT, [candidate])) + expect(() => projectTeam(ROOT, [candidate])) .toThrow(/persisted Agent Teams .* payload is invalid/) } }) it('retains merge-extensible content blocks while rejecting malformed core variants', () => { const extension = { type: 'plugin/custom', payload: { value: 1 } } as never - const state = foldTeam(ROOT, [event('team/message/queued', { + const state = projectTeam(ROOT, [event('team/message/queued', { version: 1, teamId: TEAM, message: message({ content: [extension] }), @@ -286,24 +299,26 @@ describe('Agent Teams fold', () => { expect(pending(state)[0]?.content).toEqual([extension]) }) - it('rejects unsupported event versions without mutating an empty state', () => { - const state = emptyTeamFoldState(ROOT) + it('records unsupported event versions without applying them', () => { const invalid = event('team/task', { version: 2 as 1, teamId: TEAM, task: task(), }, 0) - expect(() => { applyTeamEvent(state, invalid) }).toThrow(/unsupported Agent Teams event version 2/) - expect(isEmptyFold(state)).toBe(true) + const state = project([invalid]).teams[0]! + expect(state.failure).toMatch(/unsupported Agent Teams event version 2/) + expect(isEmptyState(state)).toBe(true) }) - it('ignores unsupported inherited Team records before decoding their version', () => { + it('isolates unsupported inherited Team records from the current Team', () => { const inherited = event('team/task', { version: 2 as 1, teamId: TeamId('ancestor'), task: task(), }, 0) - expect(isEmptyFold(foldTeam(ROOT, [inherited]))).toBe(true) + const projected = project([inherited]) + expect(projected.teams[0]?.failure).toMatch(/unsupported Agent Teams event version 2/) + expect(isEmptyState(teamState(ROOT, projected))).toBe(true) }) it('still validates complete current-version records inherited from another Team', () => { @@ -319,7 +334,7 @@ describe('Agent Teams fold', () => { task: { ...task(), subject: 42 }, }, } as unknown as SessionEvent - expect(() => foldTeam(ROOT, [inherited])) + expect(() => projectTeam(ROOT, [inherited])) .toThrow(/persisted Agent Teams team\/task payload is invalid/) }) }) diff --git a/packages/experimental/team/tests/team.spec.ts b/packages/experimental/team/tests/team.spec.ts index 5c36947a48..bb27399249 100644 --- a/packages/experimental/team/tests/team.spec.ts +++ b/packages/experimental/team/tests/team.spec.ts @@ -14,8 +14,9 @@ import SubagentService from '@deepseek-ai/dsh-subagent' import * as SubagentFork from '@deepseek-ai/dsh-subagent-fork-in-process' import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn-in-process' import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' -import TeamService, { foldTeam, TeamError, TeamId, TeamMessageId, TeamTaskId } from '../src/index.ts' +import TeamService, { TeamError, TeamId, TeamMessageId, TeamTaskId } from '../src/index.ts' import { TeamRuntimeLifecycle } from '../src/lifecycle.ts' +import { emptyTeamState, teamProjectionDefinition } from '../src/projection.ts' import type { TeamMemberSnapshot, TeamMessageSnapshot, TeamTaskSnapshot } from '../src/index.ts' const SIGNAL = new AbortController().signal @@ -26,17 +27,21 @@ afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }) }) -/** Detached durable Team read: the service exposes views, so assertions fold the Lead log. */ +/** Detached durable Team read through the same projection definition as the service. */ function durable(agent: Agent): { members: TeamMemberSnapshot[] tasks: TeamTaskSnapshot[] pendingMessages: TeamMessageSnapshot[] } { - const state = foldTeam(agent.id, agent.session.events) + let projected = teamProjectionDefinition.init() + for (const event of agent.session.events) projected = teamProjectionDefinition.apply(projected, event) + const selected = projected.teams.find(team => team.id === TeamId(agent.id)) + if (selected?.failure !== undefined) throw new Error(selected.failure) + const state = selected ?? emptyTeamState(agent.id) return { - members: [...state.members.values()], - tasks: [...state.tasks.values()], - pendingMessages: [...state.messages.values()].filter(message => !state.delivered.has(message.id)), + members: state.members, + tasks: state.tasks, + pendingMessages: state.messages.filter(message => !state.delivered.includes(message.id)), } } diff --git a/packages/experimental/team/tsconfig.json b/packages/experimental/team/tsconfig.json index d0defa8242..669a476e44 100644 --- a/packages/experimental/team/tsconfig.json +++ b/packages/experimental/team/tsconfig.json @@ -15,6 +15,7 @@ { "path": "../../core/agent" }, { "path": "../../subagent/subagent" }, { "path": "../../session/session-persistence" }, + { "path": "../../session/session-projection" }, { "path": "../../runtime-diagnostics/invariants" } ] } diff --git a/packages/experimental/team/tsdown.config.ts b/packages/experimental/team/tsdown.config.ts index e630b728b4..cc38b29b83 100644 --- a/packages/experimental/team/tsdown.config.ts +++ b/packages/experimental/team/tsdown.config.ts @@ -1,6 +1,6 @@ import { defineConfig } from 'tsdown' -/** Build the runtime and invariant as independent bundles so shared fold code stays package-local. */ +/** Build the runtime and invariant companion as independent package exports. */ export default defineConfig([ { entry: ['lib/types/index.js'], diff --git a/packages/session/session-projection/src/index.ts b/packages/session/session-projection/src/index.ts index 6029947c68..dbfc554330 100644 --- a/packages/session/session-projection/src/index.ts +++ b/packages/session/session-projection/src/index.ts @@ -263,7 +263,7 @@ export class SessionProjectionRegistry extends Service { if (live.refs === 0) this.registrations.delete(key) } }.bind(this), 'sessionProjections.register()') - return () => void dispose() + return dispose } /** diff --git a/packages/session/session-projection/tests/registry.spec.ts b/packages/session/session-projection/tests/registry.spec.ts index 459e8a1a5b..da0cb18cfb 100644 --- a/packages/session/session-projection/tests/registry.spec.ts +++ b/packages/session/session-projection/tests/registry.spec.ts @@ -273,6 +273,23 @@ describe('SessionProjectionRegistry drive', () => { expect(ctx.sessionProjections.snapshot(session).values['test/marks']).toEqual({ marks: ['cached'] }) }) + it('returns the exact disposer so a composite owner can read state before unregistering', async () => { + const { ctx, session } = await harness() + mark(session, ['cleanup']) + let cleanupState: MarksState | undefined + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + inner.effect(function* () { + yield inner.sessionProjections.register(marksUnit()) + yield () => { cleanupState = inner.sessionProjections.stateOf(session, 'test/marks') } + }, 'projection cleanup ordering fixture') + }, { inject: ['sessionProjections'] })) + + await fiber.dispose() + + expect(cleanupState).toEqual({ marks: ['cleanup'] }) + expect(ctx.sessionProjections.stateOf(session, 'test/marks')).toBeUndefined() + }) + it('removes registrations and change listeners when their owning fiber unloads (HMR safety)', async () => { const { ctx, session } = await harness() const notifications: string[] = []