diff --git a/packages/agent-core-v2/docs/Permission.md b/packages/agent-core-v2/docs/Permission.md new file mode 100644 index 000000000..8424b1df1 --- /dev/null +++ b/packages/agent-core-v2/docs/Permission.md @@ -0,0 +1,327 @@ +# 权限系统设计(Permission) + +本文系统整理 agent-core 权限系统的目标方案,并与 `packages/agent-core`(v1)现状对比。结论先行: + +> **权限系统应是一个「可组合、可注册的责任链(微内核)」**:内核只负责按顺序跑链、首个命中赢;具体权限维度(policy)由各自的 Domain Service 通过注册表插入;工具只需在 `resolveExecution` 里声明标准化的资源访问(`accesses`),通用维度集中消费这份元数据。 +> +> **不引入 Casbin**——因为这里「难的是决策行为」(续体、副作用、RPC、状态机),不是「匹配 + 标量决策」。 + +--- + +## 一、背景与问题定义 + +权限系统回答一个问题:**对于每一次工具调用,在当前 agent、当前 mode 下,放行 / 拒绝 / 询问用户?** + +这个决策有三个特点,决定了它的架构取向: + +1. **决策携带行为**。返回 `ask` 不是一个枚举值,而是一条含 RPC 往返、hook、telemetry、状态写入、续体的工作流;返回 `deny` 可能是执行了一段外部 hook 的结果。 +2. **策略异质**。有的查工具名集合,有的数同批 AgentSwarm 个数,有的跑 hook,有的检查 plan 状态机——没有统一的 `(sub, obj, act)` 形状。 +3. **多 agent × 多 mode × 外部扩展**。不同 agent / mode 需要不同权限,且要允许外部(组织管理员、插件)解耦地贡献规则或行为。 + +--- + +## 二、现状(agent-core v1) + +代码位于 `packages/agent-core/src/agent/permission/`。 + +### 2.1 架构:有序责任链 + 首个命中赢 + +`PermissionManager`(`index.ts`)持有一组 `PermissionPolicy`,决策时顺序遍历,第一个返回非 `undefined` 的 policy 胜出: + +```ts +// index.ts evaluatePolicies +for (const policy of this.policies) { + const result = await policy.evaluate(context); + if (result !== undefined) return { policyName: policy.name, result }; +} +``` + +每个 policy 是一个实现 `PermissionPolicy` 接口的类,`evaluate(context)` 不适用就返回 `undefined`(传给下一个)。`PermissionPolicyResult` 不是标量,而是可携带续体和副作用的「行为包」: + +```ts +// types.ts +type PermissionPolicyResult = + | { kind: 'approve'; reason?; executionMetadata? } + | { kind: 'deny'; reason?; message? } + | { kind: 'ask'; reason?; resolveApproval?; resolveError? }; +``` + +### 2.2 11 个权限维度(19 个 policy) + +链目前在 `policies/index.ts#createPermissionDecisionPolicies()` 中**硬编码**,顺序即优先级。19 个 policy 可归并为 11 个权限维度: + +| # | 维度 | 对应 policy | 决策看什么 | +|---|---|---|---| +| 1 | 外部钩子否决 | `pre-tool-call-hook` | 用户 `PreToolUse` hook 是否返回 block | +| 2 | 工具批量排他 | `agent-swarm-exclusive-deny`、`swarm-mode-agent-swarm-approve` | 同批工具结构(AgentSwarm 须单独)+ swarm 模式 | +| 3 | 运行模式姿态 | `auto-mode-approve`、`yolo-mode-approve`、`auto-mode-ask-user-question-deny` | `permission.mode` | +| 4 | Plan 模式约束 | `plan-mode-guard-deny`、`plan-mode-tool-approve`、`exit-plan-mode-review-ask` | `planMode.isActive` + plan 文件路径 + review 状态 | +| 5 | Goal 启动审批 | `goal-start-review-ask` | `tool === CreateGoal` 且非 auto | +| 6 | 静态配置规则 | `user-configured-deny/ask/allow` | 用户/项目/turn 配置的 DSL 规则 | +| 7 | 会话批准记忆 | `session-approval-history` | 本会话 "approve for session" 缓存 | +| 8 | 敏感/特殊路径 | `sensitive-file-access-ask`、`git-control-path-access-ask` | 工具访问的文件路径 | +| 9 | 工具内在风险 | `default-tool-approve` | 工具名 ∈ 默认安全集合 | +| 10 | 工作区写信任 | `git-cwd-write-approve` | POSIX + git worktree + cwd 内写 | +| 11 | 兜底 | `fallback-ask` | 无(默认 ask) | + +链的顺序是一条**从高到低的安全级联**:外部强制 → 结构性拒绝 → 状态机拒绝 → 静态 deny → mode 放行 → 会话记忆放行 → 静态 ask → 静态 allow → 流程放行 → 敏感路径 ask → 默认放行 → 兜底 ask。 + +### 2.3 资源访问声明:`resolveExecution` + `accesses` + +工具通过 `resolveExecution(input)` 在执行前声明自己访问的资源(`packages/agent-core/src/loop/types.ts`、`tool-access.ts`): + +```ts +interface RunnableToolExecution { + readonly accesses?: ToolAccesses; // 资源 + 操作 + readonly matchesRule?: (ruleArgs) => boolean; + readonly approvalRule: string; + readonly execute: (ctx) => Promise; +} +``` + +`ToolAccesses` 是 `ToolResourceAccess[]`,目前支持 `file` 与 `all` 两类资源(详见 §5.5)。权限维度(如 `sensitive-file-access-ask`、`git-cwd-write-approve`)读 `context.execution.accesses` 做判断。 + +### 2.4 优势 + +- **清晰可审计**:顺序显式,每个 policy 旁有注释解释其位置,安全姿态一目了然。 +- **首个命中短路**:大多数调用(如只读工具)在 `default-tool-approve` 即返回,性能好。 +- **行为表达力强**:`ask` 可携带 `resolveApproval` 续体、`executionMetadata`、自定义消息和副作用。 + +### 2.5 痛点 + +1. **链硬编码**。19 个 policy 在一个函数里 `new`,外部无法贡献。 +2. **mode 是 policy 内部的 `if`**。`YoloModeApprove` / `AutoModeApprove` 各自 `if (mode !== 'x') return`,"不同 mode 不同链"只能靠塞更多 self-guard 的 policy。 +3. **没有按 agent 区分链的入口**(只有散落的 `agent.type === 'sub'` 判断)。 +4. **没有外部扩展点**。唯一的外部介入是 `PreToolUse` hook(占 guard 一个固定槽位)。 +5. **bash/write 等通用工具的维度集中在核心**,工具自己只声明 `accesses`,不知道维度存在——这是优点,但也意味着新增维度要改核心。 + +--- + +## 三、为什么不是 Casbin + +Casbin 的两个卖点(`policy_effect` 和灵活 priority)在当前业务下都落不到实处。 + +### 3.1 `policy_effect` 用不上 + +`policy_effect` 解决「多规则命中后如何组合」。但 agent-core 的组合逻辑是**固定的安全级联**,且真正的复杂度在每条 policy 的 `evaluate` 行为里,Casbin 表达式吸收不了。更重要的是:组合顺序是安全相关的、故意写死的姿态,不希望外部改动——外部可调的安全旋钮已通过 `mode` + allow/deny/ask 规则暴露。 + +### 3.2 灵活 priority 用不上 + +priority 的痛点是「多模块各自贡献规则时数字撞车」。agent-core 当前没有插件注入点、没有多主体/RBAC,主体固定(agent/用户),不存在撞车问题。Casbin 的 `(sub, obj, act)`、`g()`、domain 等抽象在这里空转。 + +### 3.3 根本性不匹配:决策不是标量 + +`enforce()` 的契约是「输入请求 → 输出 effect」。agent-core 的决策是**行为包**: + +| policy | 返回 `ask` 后的真实行为 | +|---|---| +| `requestToolApproval` | 触发 hook → 异步 RPC 问用户 → 记 telemetry → 写 records/replay → 可选写会话缓存 → 调续体 | +| `goal-start-review-ask` | 弹菜单 → 根据回答**切换 permission mode** → 放行 | +| `exit-plan-mode-review-ask` | 推进 plan 状态机 → 记多种 telemetry → **合成工具结果**短路执行 | +| `pre-tool-call-hook` | `deny` 是**异步执行外部 hook** 的结果 | + +这些续体、副作用、合成结果没有槽位放进 Casbin 的标量 effect。即便让 Casbin 算出 `ask`,外面仍需重写一整套把 `ask` 关联到行为的逻辑——Casbin 降级成枚举生成器。 + +### 3.4 Casbin 何时才值得 + +当「难的是匹配语义本身」时——角色继承、domain 隔离、ABAC 表达式、从 DB 加载策略——Casbin 才有用武之地。在此之前不引入。 + +--- + +## 四、设计模式定位 + +权限编排不是一个单一模式,而是分层组合: + +| 层 | 模式 | 作用 | +|---|---|---| +| 运行时决策 | **责任链(Chain of Responsibility)** | 多个候选处理者按顺序,首个命中赢,后续短路 | +| 单个处理者 | **策略(Strategy)** | 每个 policy 是「权限裁决」算法族的可互换实现 | +| 组装 / 外部扩展 | **插件 / 微内核(Plugin / Microkernel)** | 极简内核 + 明确扩展点 + 可插拔的 policy | +| 落地辅助 | **注册表(Registry)+ 工厂(Factory)** | 收集插件;按 (agent, mode) 现场组装链 | + +与 Casbin 的范式对比: + +- **Casbin = 单一 Strategy + 数据驱动**:所有决策走同一个 matcher 表达式,差异压成 policy rows(数据)。 +- **本方案 = 多 Strategy + 责任链组合**:每个 policy 是独立策略,差异靠代码,靠责任链组装。 + +行为密集型系统必须选后者——行为无法压成数据行。 + +--- + +## 五、目标方案 + +### 5.1 核心原则 + +1. **链编码「权限维度」,不编码「工具」**。新增工具不延长链;只有新增维度才加节点。 +2. **两条贡献路径**:高频琐碎的具体内容走**数据路径**(规则);低频有行为的新维度走**代码路径**(policy)。 +3. **Domain 自注册**:拥有专属维度的 domain(plan/goal/swarm)在 DI 中自注册 policy,镜像 v2 已有的「domain 自注册工具」。 +4. **工具声明资源,通用维度消费**:bash/write/read 等只声明 `accesses`,文件/安全维度集中判断。 + +### 5.2 核心抽象 + +```ts +type Phase = + | 'guard' | 'user-deny' | 'mode' | 'session' + | 'user-ask' | 'default' | 'fallback'; + +interface PermissionPolicyEntry { + name: string; + phase: Phase; + modes?: PermissionMode[]; // 声明在哪些 mode 生效(不再在 evaluate 里 if) + agentTypes?: AgentType[]; + factory: (accessor: ServicesAccessor) => PermissionPolicy; +} + +// Core scope —— 收集所有 domain 的注册 +interface IPermissionPolicyRegistry { + register(entry: PermissionPolicyEntry): IDisposable; + list(): readonly PermissionPolicyEntry[]; +} +``` + +`PermissionPolicyService`(Agent scope)从硬编码列表改为「按 (agent, mode) 组装」: + +```ts +this.policies = registry.list() + .filter(e => !e.modes || e.modes.includes(mode)) + .filter(e => !e.agentTypes || e.agentTypes.includes(agentType)) + .sort(byPhaseThenRegistrationOrder) + .map(e => e.factory(accessor)); +``` + +要点: + +- `modes`/`agentTypes` 是**声明**,把现在 `YoloModeApprove` 里的 `if (mode !== 'yolo') return` 提到元数据。 +- `factory` 而非 `instance`:节点可能依赖 agent-scoped 服务(mode、rules),需在 Agent scope 实例化——对称 `IToolDefinitionRegistry`(Core) 存 factory、`IToolService`(Agent) 实例化工具。 +- **不同 (agent, mode) 产出形状不同的链**:yolo 下 ask/fallback 阶段被物理过滤掉。 + +### 5.3 两条贡献路径 + +| 新增的是…… | 路径 | 链长变化 | +|---|---|---| +| 新工具、新组织规则、新用户偏好("禁 `Bash(curl *)`") | **数据路径**:往现有节点塞一条 `PermissionRule` | 不变 | +| 新横切行为(自定义审批 UI、审计日志、新 mode) | **代码路径**:注册一个新 policy 节点 | +1 | + +绝大部分增长走数据路径——节点数被「行为种类」约束,规则数才随具体情况增长(规则匹配是廉价的 Set/glob)。 + +### 5.4 Domain 自注册 + +镜像 v2 里「domain 在构造函数中 `toolRegistry.register(...)`」的现成做法。PlanService 自注册其维度: + +```ts +// src/plan/planService.ts +constructor(@IPermissionPolicyRegistry registry: IPermissionPolicyRegistry) { + registry.register({ name: 'plan-mode-guard-deny', phase: 'guard', + factory: a => new PlanModeGuardDenyPolicy(a.get(IPlanService)) }); + registry.register({ name: 'plan-mode-tool-approve', phase: 'mode', + factory: a => new PlanModeToolApprovePolicy(a.get(IPlanService)) }); + registry.register({ name: 'exit-plan-mode-review-ask', phase: 'user-ask', + factory: a => new ExitPlanModeReviewAskPolicy(a.get(IPlanService), a.get(IPermissionModeService)) }); +} +``` + +复杂 domain 可对外只注册**一个复合节点**(Composite),内部跑小链,避免泄漏内部顺序到全局。 + +### 5.5 工具运行时声明资源(`resolveExecution` / `accesses`) + +工具在 `resolveExecution(input)` 里、执行前,用 `ToolAccesses.*` builder 声明访问的资源: + +```ts +// packages/agent-core/src/tools/builtin/file/write.ts +resolveExecution(args: WriteInput): ToolExecution { + const path = resolvePathAccessPath(args.path, { kaos, workspace, operation: 'write' }); + return { + accesses: ToolAccesses.writeFile(path), // 声明:写这个文件 + approvalRule: literalRulePattern(this.name, path), + matchesRule: (ruleArgs) => matchesPathRuleSubject(ruleArgs, path, ...), + execute: () => this.execution(args, path), + }; +} +``` + +`ToolAccesses` 目前两类资源: + +```ts +type ToolResourceAccess = + | { kind: 'file'; operation: 'read'|'write'|'readwrite'|'search'; path: string; recursive?: boolean } + | { kind: 'all' }; // 无法枚举的副作用(悲观、全局排他) +``` + +**两条互补通道**: + +- **能枚举资源的**(write/read/edit/grep/glob)→ 用 `accesses`,通用文件维度自动覆盖。 +- **不能枚举资源的**(bash 跑任意命令)→ 不声明 `accesses`,改用 `matchesRule` DSL(如 `Bash(rm *)` 按命令串 glob)。 + +**kaos 的定位**:kaos 是执行环境抽象(fs/process/pathClass),供文件维度做路径归一化与判断,**不是权限维度抽象本身**。权限语义在 kaos 之上的「文件访问」层。 + +**v2 演进方向**:扩展 `ToolResourceAccess` 联合类型,让非文件资源也能结构化声明: + +```ts +type ToolResourceAccess = + | { kind: 'file'; operation: FileOp; path: string; recursive?: boolean } + | { kind: 'network'; operation: 'connect'; host: string } + | { kind: 'shell'; command: string } + | { kind: 'datastore'; operation: 'read'|'write'; table: string } + | { kind: 'all' }; +``` + +每新增一种资源类型,可对应加一个通用维度消费它;工具侧始终只负责**声明**。 + +### 5.6 维度归属 + +| 维度 | 拥有者(谁注册) | 类型 | +|---|---|---| +| 外部钩子否决 | `externalHooks` domain | 通用 | +| 工具批量排他 | `swarm` domain | domain 专属(跟 AgentSwarm 工具一起走) | +| 运行模式姿态 | `permissionMode` domain | 通用 | +| Plan 模式约束 | `plan` domain | domain 专属 | +| Goal 启动审批 | `goal` domain | domain 专属 | +| 静态配置规则 | `permissionRules` domain | 通用(数据路径) | +| 会话批准记忆 | `permissionRules` domain | 通用 | +| 敏感/特殊路径 | 通用「文件访问/安全」维度 | 通用(消费 `accesses`) | +| 工具内在风险 | 核心 permission | 通用(消费工具声明) | +| 工作区写信任 | 通用「文件访问/安全」维度 | 通用(消费 `accesses`) | +| 兜底 | 核心 permission | 通用 | + +规律:**专属维度跟着拥有它的 domain + 工具一起走;通用维度集中注册,靠工具声明的 `accesses` 跨工具生效。** + +--- + +## 六、现状 vs 方案 对比 + +| 方面 | 现状(v1) | 目标方案 | +|---|---|---| +| 链的构造 | `policies/index.ts` 硬编码 19 个 `new` | `IPermissionPolicyRegistry` 收集,`compose(agent, mode)` 组装 | +| mode 处理 | policy 内部 `if (mode !== 'x') return` | 声明式 `modes` 元数据,compose 时过滤 | +| 按 agent 区分 | 散落 `agent.type === 'sub'` | 声明式 `agentTypes` 元数据 | +| 外部扩展 | 仅 `PreToolUse` hook 一个固定槽 | 注册表开放注册 policy(代码)+ rule(数据) | +| Domain 维度 | 集中在核心文件 | plan/goal/swarm 各自 domain 自注册 | +| 工具维度 | 工具声明 `accesses`,维度集中 | 不变,扩展 `ToolResourceAccess` 资源类型 | +| 决策行为 | 续体 + 副作用(已具备) | 不变(这是必须保留的核心能力) | +| 运行时性能 | 顺序链 + 短路 | 不变;节点增多时可加工具名索引优化 | + +**不变的**:责任链内核、首个命中赢、`PermissionPolicyResult` 行为包、`resolveExecution`/`accesses` 机制。 + +**改变的**:链从「硬编码列表」变成「注册表 + 工厂组装」;mode/agent 从「内部 if」变成「声明式元数据」;维度归属从「核心集中」变成「domain 自注册」。 + +--- + +## 七、演进路径 + +渐进式,避免一步到位: + +1. **第一步:注册表 + Composer(行为零变化)**。把 v2 `PermissionPolicyService` 构造函数里硬编码的 19 个 `new`,改为从 `IPermissionPolicyRegistry` 读取并组装;现有 policy 原样注册。立刻获得多 agent/mode 可选链与外部注册入口。 +2. **第二步:声明式 modes**。把 `YoloModeApprove` / `AutoModeApprove` 里的 mode 守门提到 `modes` 元数据。 +3. **第三步:Domain 维度下沉**。把 plan/goal/swarm 相关 policy 的注册移到各自 domain service 构造函数。 +4. **第四步(按需):扩展资源类型**。当非文件资源(网络/DB/shell)需要结构化维度时,扩展 `ToolResourceAccess` 联合。 +5. **第五步(按需):匹配内核换 Casbin**。仅当外部规则真的需要 RBAC/ABAC 语义时,把数据路径的规则匹配内核换成 Casbin。不到此步不引。 + +--- + +## 八、待决问题 + +1. **Composite 节点的边界**:哪些 domain 内部用复合节点(隐藏子顺序),哪些直接注册多个 phase 节点? +2. **同 phase 多节点的排序**:注册顺序是否足够,还是需要显式 `order` 逃生舱? +3. **`ToolResourceAccess` 扩展节奏**:哪些非文件资源优先纳入(shell / network / datastore)? +4. **v1 → v2 迁移时机**:v2 权限子系统目前是 v1 类型/逻辑的薄包装,何时把 `accesses`、`PermissionPolicyResult` 等提升为正式 v2 类型? +5. **运行时性能阈值**:节点数达到多少时引入工具名索引(`byTool` 分派)优化?当前 19 个节点、首个命中短路,远未触及。 diff --git a/packages/agent-core-v2/docs/di-scope-domains.puml b/packages/agent-core-v2/docs/di-scope-domains.puml index 6303528a9..02adccf3d 100644 --- a/packages/agent-core-v2/docs/di-scope-domains.puml +++ b/packages/agent-core-v2/docs/di-scope-domains.puml @@ -86,7 +86,7 @@ package "Turn scope (per turn)" #FDEDEC { } package "Multi-scope" #F5EEF8 { - rectangle "config\nMulti\n IConfigRegistry(C)\n IConfigService(C)\n IAgentConfigService(A)" as config #E8DAEF + rectangle "config\nMulti\n IConfigRegistry(C)\n IConfigService(C)\n ISessionConfigService(S)" as config #E8DAEF rectangle "kosong\nMulti\n IModelCatalogService(C)\n IProviderManager(S)\n ILLMService(A)" as kosong #E8DAEF rectangle "records\nMulti\n ISessionStore(C)\n ISessionMetaStore(S)\n IAgentRecords(A)" as records #E8DAEF rectangle "tool\nMulti\n IToolDefinitionRegistry(C)\n IToolService(A)" as tool #E8DAEF diff --git a/packages/agent-core-v2/docs/di-scope-domains.svg b/packages/agent-core-v2/docs/di-scope-domains.svg index 2a3d324be..4fbda2041 100644 --- a/packages/agent-core-v2/docs/di-scope-domains.svg +++ b/packages/agent-core-v2/docs/di-scope-domains.svg @@ -1 +1 @@ -Core scope (process-wide)Session scope (per session)Agent scope (per agent)Turn scope (per turn)Multi-scopeWIP (contract only)environmentCoreIEnvironmentServicelogCoreILogServiceILogSinktelemetryCoreITelemetryServiceeventCoreIEventServicegatewayCoreIScopeRegistryIRestGatewayIWSGatewayIWSBroadcastServiceworkspaceCoreIWorkspaceRegistryIWorkspaceFsServicefilestoreCoreIFileStoreauthCoreIOAuthServiceIAuthSummaryServiceflagCoreIFlagServicesessionSessionISessionServicefsSessionIFsServiceIFsSearchServiceIFsGitServiceIFsWatcherapprovalSessionIApprovalServicequestionSessionIQuestionServicesession-activitySessionISessionActivitysubagentHostSessionISubagentHostterminalSessionITerminalServiceagent-lifecycleSessionIAgentLifecycleServicesession-contextSessionISessionContext (seed)permissionModeAgentIPermissionModeServicepromptAgentIPromptServiceuserToolAgentIUserToolServicellmRequestLogAgentILLMRequestLogServicedynamicInjectorAgentIDynamicInjectorreplayBuilderAgentIReplayBuilderServiceprofileAgentIProfileServicerpcAgentIAgentRPCServicecontextMemoryAgentIContextMemorytoolExecutorAgentIToolExecutorpermissionAgentIPermissionServicecronAgentICronServicellmRequesterAgentILLMRequesterfullCompactionAgentIFullCompactionpermissionRulesAgentIPermissionRulesServiceskillAgentIAgentSkillServicegoalAgentIGoalServicemicroCompactionAgentIMicroCompactionServiceexternalHooksAgentIExternalHooksServiceswarmAgentISwarmServicebackgroundAgentIBackgroundServicetoolStoreAgentIToolStoreServicewireRecordAgentIWireRecord (event hub)eventBusAgentIEventBustoolRegistryAgentIToolRegistryusageAgentIUsageServicemessageAgentIMessageServicemcpAgentIMcpServicecontextSizeAgentIContextSizeServicepermissionPolicyAgentIPermissionPolicyServicecontextProjectorAgentIContextProjectortodoListAgentITodoListServiceloopAgentILoopServiceturnAgentITurnServiceITurnEventsILoopRunner(T)IToolCallExecutor(T)planAgentIPlanServiceblobStoreAgentIBlobStoreServicetooldedupTurnIToolDedupServiceconfigMultiIConfigRegistry(C)IConfigService(C)IAgentConfigService(A)kosongMultiIModelCatalogService(C)IProviderManager(S)ILLMService(A)recordsMultiISessionStore(C)ISessionMetaStore(S)IAgentRecords(A)toolMultiIToolDefinitionRegistry(C)IToolService(A)kaosMultiIKaosFactory(C)ISessionKaosService(S)IKaosService / IAgentKaos(A)contextWIPIContextService (WIP)injectionWIPIInjectionService (WIP)turnRunnerWIPITurnRunner (WIP)subscribeonDidChangepermission.set_modetools.register_/unregister_user_toolconfig.update / tools.set_active_toolscontext.splicecron.add / delete / cursorfull_compaction.begin/cancel/completepermission.rules.add / record_approval_resultskill.activategoal.create/update/clearmicro_compaction.apply / full_compaction.completeswarm_mode.enter/exithooks.onEndedbackground.task.started/terminatedhooks.onSplicedtools.update_storeusage.recordhooks.onSplicedcontext_size.measuredhooks.onSplicedhooks.onResumeEndedonWillExecuteToolplan_mode.enter/cancel/exitScope = node color      Core (process-wide, L0)      Session (per session, L1)      Agent (per agent, L2)      Turn (per turn, L3)      Multi-scope (spans layers)      WIP: contract onlyEdgessolid: DI injection (ctor @IX)dashed: event-driven (subscribe/emit)direction: consumer ---> provider \ No newline at end of file +Core scope (process-wide)Session scope (per session)Agent scope (per agent)Turn scope (per turn)Multi-scopeWIP (contract only)environmentCoreIEnvironmentServicelogCoreILogServiceILogSinktelemetryCoreITelemetryServiceeventCoreIEventServicegatewayCoreIScopeRegistryIRestGatewayIWSGatewayIWSBroadcastServiceworkspaceCoreIWorkspaceRegistryIWorkspaceFsServicefilestoreCoreIFileStoreauthCoreIOAuthServiceIAuthSummaryServiceflagCoreIFlagServicesessionSessionISessionServicefsSessionIFsServiceIFsSearchServiceIFsGitServiceIFsWatcherapprovalSessionIApprovalServicequestionSessionIQuestionServicesession-activitySessionISessionActivitysubagentHostSessionISubagentHostterminalSessionITerminalServiceagent-lifecycleSessionIAgentLifecycleServicesession-contextSessionISessionContext (seed)permissionModeAgentIPermissionModeServicepromptAgentIPromptServiceuserToolAgentIUserToolServicellmRequestLogAgentILLMRequestLogServicedynamicInjectorAgentIDynamicInjectorreplayBuilderAgentIReplayBuilderServiceprofileAgentIProfileServicerpcAgentIAgentRPCServicecontextMemoryAgentIContextMemorytoolExecutorAgentIToolExecutorpermissionAgentIPermissionServicecronAgentICronServicellmRequesterAgentILLMRequesterfullCompactionAgentIFullCompactionpermissionRulesAgentIPermissionRulesServiceskillAgentIAgentSkillServicegoalAgentIGoalServicemicroCompactionAgentIMicroCompactionServiceexternalHooksAgentIExternalHooksServiceswarmAgentISwarmServicebackgroundAgentIBackgroundServicetoolStoreAgentIToolStoreServicewireRecordAgentIWireRecord (event hub)eventBusAgentIEventBustoolRegistryAgentIToolRegistryusageAgentIUsageServicemessageAgentIMessageServicemcpAgentIMcpServicecontextSizeAgentIContextSizeServicepermissionPolicyAgentIPermissionPolicyServicecontextProjectorAgentIContextProjectortodoListAgentITodoListServiceloopAgentILoopServiceturnAgentITurnServiceITurnEventsILoopRunner(T)IToolCallExecutor(T)planAgentIPlanServiceblobStoreAgentIBlobStoreServicetooldedupTurnIToolDedupServiceconfigMultiIConfigRegistry(C)IConfigService(C)ISessionConfigService(S)kosongMultiIModelCatalogService(C)IProviderManager(S)ILLMService(A)recordsMultiISessionStore(C)ISessionMetaStore(S)IAgentRecords(A)toolMultiIToolDefinitionRegistry(C)IToolService(A)kaosMultiIKaosFactory(C)ISessionKaosService(S)IKaosService / IAgentKaos(A)contextWIPIContextService (WIP)injectionWIPIInjectionService (WIP)turnRunnerWIPITurnRunner (WIP)subscribeonDidChangepermission.set_modetools.register_/unregister_user_toolconfig.update / tools.set_active_toolscontext.splicecron.add / delete / cursorfull_compaction.begin/cancel/completepermission.rules.add / record_approval_resultskill.activategoal.create/update/clearmicro_compaction.apply / full_compaction.completeswarm_mode.enter/exithooks.onEndedbackground.task.started/terminatedhooks.onSplicedtools.update_storeusage.recordhooks.onSplicedcontext_size.measuredhooks.onSplicedhooks.onResumeEndedonWillExecuteToolplan_mode.enter/cancel/exitScope = node color      Core (process-wide, L0)      Session (per session, L1)      Agent (per agent, L2)      Turn (per turn, L3)      Multi-scope (spans layers)      WIP: contract onlyEdgessolid: DI injection (ctor @IX)dashed: event-driven (subscribe/emit)direction: consumer ---> provider \ No newline at end of file diff --git a/packages/agent-core-v2/docs/service-design.md b/packages/agent-core-v2/docs/service-design.md index 1afe1f36a..55d72068c 100644 --- a/packages/agent-core-v2/docs/service-design.md +++ b/packages/agent-core-v2/docs/service-design.md @@ -118,7 +118,7 @@ This pattern recurs throughout the codebase and confirms the rule: (`Session`, this session's metadata) + `IAgentRecords` (`Agent`, this agent's record stream). - **`config`** — `IConfigRegistry` / `IConfigService` (`Core`, global config) + - `IAgentConfigService` (`Agent`, this agent's config view). + `ISessionConfigService` (`Session`, this session's runtime config). - **`kosong`** — `IModelCatalogService` (`Core`, model catalog) + `IProviderManager` (`Session`) + `ILLMService` (`Agent`, this agent's generation). - **`tool`** — `IToolDefinitionRegistry` (`Core`, tool-definition registry) + `IToolService` diff --git a/packages/agent-core-v2/scripts/check-domain-layers.mjs b/packages/agent-core-v2/scripts/check-domain-layers.mjs index 529bcd490..c5c41a101 100644 --- a/packages/agent-core-v2/scripts/check-domain-layers.mjs +++ b/packages/agent-core-v2/scripts/check-domain-layers.mjs @@ -43,6 +43,8 @@ const DOMAIN_LAYER = new Map([ ['kosong', 1], // L2 — data ['records', 2], + ['wireRecord', 2], + ['blobStore', 2], ['config', 2], // L3 — registries ['tool', 3], @@ -117,6 +119,7 @@ const ALLOWED_EXCEPTIONS = new Set([ 'cron>session-context', 'cron>session-activity', 'session>event', + 'wireRecord>hooks', ]); // Matches: import ... from 'x' | export ... from 'x' | import('x') | require('x') diff --git a/packages/agent-core-v2/src/_base/errors/codes.ts b/packages/agent-core-v2/src/_base/errors/codes.ts index a29561415..1ab988ce6 100644 --- a/packages/agent-core-v2/src/_base/errors/codes.ts +++ b/packages/agent-core-v2/src/_base/errors/codes.ts @@ -11,6 +11,7 @@ export const ErrorCodes = { CONTEXT_OVERFLOW: 'context.overflow', PROVIDER_RATE_LIMIT: 'provider.rate_limit', PROVIDER_AUTH_ERROR: 'provider.auth_error', + AUTH_LOGIN_REQUIRED: 'auth.login_required', } as const; export type ErrorCode = (typeof ErrorCodes)[keyof typeof ErrorCodes]; @@ -65,6 +66,12 @@ export const ERROR_INFO = { public: true, action: 'Check provider credentials and authentication configuration.', }, + 'auth.login_required': { + title: 'Login required', + retryable: false, + public: true, + action: 'Run /login to authenticate with the OAuth provider.', + }, } as const satisfies Record; export function errorInfo(code: ErrorCode): ErrorInfo { diff --git a/packages/agent-core-v2/src/auth/auth.ts b/packages/agent-core-v2/src/auth/auth.ts index cb36b870f..d4a781c5b 100644 --- a/packages/agent-core-v2/src/auth/auth.ts +++ b/packages/agent-core-v2/src/auth/auth.ts @@ -1,14 +1,25 @@ /** - * `auth` domain (cross-cutting) — core-scope OAuth + auth summary. + * `auth` domain (cross-cutting) — core-scope OAuth + auth summary contracts. * * Defines the public contracts of authentication: the `AuthStatus` model, the - * `IOAuthService` used to log in/out and query status, and the - * `IAuthSummaryService` used to summarize auth state. Core-scoped — shared - * across the application. + * `IOAuthService` used to drive device-code login / logout / flow inspection + * and to resolve a per-provider `BearerTokenProvider`, and the + * `IAuthSummaryService` used to summarize auth state and assert readiness. + * Core-scoped — shared across the application. */ +import type { BearerTokenProvider } from '@moonshot-ai/kimi-code-oauth'; +import type { + OAuthFlowSnapshot, + OAuthFlowStart, + OAuthLoginCancelResponse, + OAuthLogoutResponse, +} from '@moonshot-ai/protocol'; + import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { OAuthRef } from './oauthSchemas'; + export interface AuthStatus { readonly loggedIn: boolean; readonly provider?: string; @@ -16,9 +27,12 @@ export interface AuthStatus { export interface IOAuthService { readonly _serviceBrand: undefined; - login(provider: string): Promise; - logout(provider: string): Promise; - status(): Promise; + startLogin(provider?: string): Promise; + getFlow(provider?: string): OAuthFlowSnapshot | undefined; + cancelLogin(provider?: string): Promise; + logout(provider?: string): Promise; + status(provider?: string): Promise; + resolveTokenProvider(provider: string, oauthRef?: OAuthRef): BearerTokenProvider | undefined; } export const IOAuthService: ServiceIdentifier = @@ -27,6 +41,7 @@ export const IOAuthService: ServiceIdentifier = export interface IAuthSummaryService { readonly _serviceBrand: undefined; summarize(): Promise; + ensureReady(provider?: string): Promise; } export const IAuthSummaryService: ServiceIdentifier = diff --git a/packages/agent-core-v2/src/auth/authService.ts b/packages/agent-core-v2/src/auth/authService.ts index a871d7647..33af5bc0d 100644 --- a/packages/agent-core-v2/src/auth/authService.ts +++ b/packages/agent-core-v2/src/auth/authService.ts @@ -2,58 +2,283 @@ * `auth` domain (cross-cutting) — `IOAuthService` / `IAuthSummaryService` * implementation. * - * Owns the OAuth login state and auth summary; reads settings through `config`, - * reads the environment through `environment`, and reports through `telemetry`. + * Owns the device-code OAuth flows and the auth readiness view; reads the + * `providers` config section through `config`, locates token storage through + * `environment`, reports through `telemetry`, and delegates token storage, + * refresh, and the device-code protocol to `@moonshot-ai/kimi-code-oauth`. * Bound at Core scope. */ +import { randomUUID } from 'node:crypto'; + +import { + DeviceCodeTimeoutError, + KIMI_CODE_PROVIDER_NAME, + KimiOAuthToolkit, + OAuthError, + type BearerTokenProvider, + type DeviceAuthorization, +} from '@moonshot-ai/kimi-code-oauth'; +import type { + OAuthFlowSnapshot, + OAuthFlowStart, + OAuthFlowStatus, + OAuthLoginCancelResponse, + OAuthLogoutResponse, +} from '@moonshot-ai/protocol'; + import { InstantiationType } from '#/_base/di/extensions'; +import { Disposable } from '#/_base/di/lifecycle'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { IConfigService } from '#/config'; -import { IEnvironmentService } from '#/environment'; -import { ITelemetryService } from '#/telemetry'; +import { ErrorCodes } from '#/_base/errors/codes'; +import { KimiError } from '#/_base/errors/errors'; +import { IConfigRegistry, IConfigService } from '#/config/config'; +import { IEnvironmentService } from '#/environment/environment'; +import { ITelemetryService } from '#/telemetry/telemetry'; import { type AuthStatus, IAuthSummaryService, IOAuthService } from './auth'; +import { + type OAuthRef, + type ProvidersSection, + PROVIDERS_SECTION, + ProvidersSectionSchema, +} from './oauthSchemas'; -export class OAuthService implements IOAuthService { +const TERMINAL_RETENTION_MS = 5 * 60 * 1000; +const DEFAULT_DEVICE_EXPIRES_IN_SEC = 15 * 60; + +interface FlowState { + readonly flowId: string; + readonly provider: string; + readonly controller: AbortController; + device: DeviceAuthorization | undefined; + status: OAuthFlowStatus; + expiresAt: number; + gcTimer: ReturnType | undefined; + errorMessage: string | undefined; + resolvedAt: string | undefined; +} + +export class OAuthService extends Disposable implements IOAuthService { declare readonly _serviceBrand: undefined; - private readonly loggedIn = new Set(); + private readonly toolkit: KimiOAuthToolkit; + private readonly flows = new Map(); constructor( - @IConfigService _config: IConfigService, - @IEnvironmentService _env: IEnvironmentService, - @ITelemetryService _telemetry: ITelemetryService, - ) {} - - login(provider: string): Promise { - this.loggedIn.add(provider); - return Promise.resolve(); - } - logout(provider: string): Promise { - this.loggedIn.delete(provider); - return Promise.resolve(); - } - status(): Promise { - const [provider] = this.loggedIn; - return Promise.resolve( - provider === undefined ? { loggedIn: false } : { loggedIn: true, provider }, + toolkit: KimiOAuthToolkit | undefined = undefined, + @IConfigRegistry registry: IConfigRegistry, + @IConfigService private readonly config: IConfigService, + @IEnvironmentService env: IEnvironmentService, + @ITelemetryService private readonly telemetry: ITelemetryService, + ) { + super(); + registry.registerSection(PROVIDERS_SECTION, ProvidersSectionSchema, { defaultValue: {} }); + this.toolkit = toolkit ?? new KimiOAuthToolkit({ homeDir: env.homeDir }); + this._register( + config.onDidChange((e) => { + if (e.domain === PROVIDERS_SECTION) { + this.invalidateFlows(); + } + }), ); } + + async startLogin(provider = KIMI_CODE_PROVIDER_NAME): Promise { + const oauthRef = this.readOAuthRef(provider); + this.abortExisting(provider); + + const state: FlowState = { + flowId: `oauth_${randomUUID()}`, + provider, + controller: new AbortController(), + device: undefined, + status: 'pending', + expiresAt: Date.now() + DEFAULT_DEVICE_EXPIRES_IN_SEC * 1000, + gcTimer: undefined, + errorMessage: undefined, + resolvedAt: undefined, + }; + this.flows.set(provider, state); + + let resolveDevice!: (auth: DeviceAuthorization) => void; + const deviceReady = new Promise((resolve) => { + resolveDevice = resolve; + }); + + const loginPromise = this.toolkit.login(provider, { + signal: state.controller.signal, + oauthRef, + onDeviceCode: (auth) => { + state.device = auth; + if (auth.expiresIn !== null) { + state.expiresAt = Date.now() + auth.expiresIn * 1000; + } + resolveDevice(auth); + }, + }); + loginPromise.then( + () => this.handleSuccess(state), + (error) => this.handleFailure(state, error), + ); + + const device = await deviceReady; + return this.toFlowStart(state, device); + } + + getFlow(provider = KIMI_CODE_PROVIDER_NAME): OAuthFlowSnapshot | undefined { + const state = this.flows.get(provider); + if (state === undefined || state.device === undefined) return undefined; + return this.toSnapshot(state, state.device); + } + + cancelLogin(provider = KIMI_CODE_PROVIDER_NAME): Promise { + const state = this.flows.get(provider); + if (state === undefined || state.status !== 'pending') { + return Promise.resolve({ cancelled: false, status: state?.status ?? 'cancelled' }); + } + state.controller.abort(); + this.setTerminal(state, 'cancelled'); + return Promise.resolve({ cancelled: true, status: 'cancelled' }); + } + + async logout(provider = KIMI_CODE_PROVIDER_NAME): Promise { + const oauthRef = this.readOAuthRefOptional(provider); + await this.toolkit.logout(provider, oauthRef); + this.abortExisting(provider); + return { logged_out: true, provider }; + } + + async status(provider = KIMI_CODE_PROVIDER_NAME): Promise { + const oauthRef = this.readOAuthRefOptional(provider); + const token = await this.toolkit.getCachedAccessToken(provider, oauthRef); + return token === undefined ? { loggedIn: false } : { loggedIn: true, provider }; + } + + resolveTokenProvider(provider: string, oauthRef?: OAuthRef): BearerTokenProvider | undefined { + return this.toolkit.tokenProvider(provider, oauthRef); + } + + private readOAuthRef(provider: string): OAuthRef { + const providers = this.config.get(PROVIDERS_SECTION); + const oauth = providers?.[provider]?.oauth; + if (oauth === undefined) { + throw new KimiError( + ErrorCodes.AUTH_LOGIN_REQUIRED, + `Provider "${provider}" is not configured for OAuth.`, + ); + } + return oauth; + } + + private readOAuthRefOptional(provider: string): OAuthRef | undefined { + const providers = this.config.get(PROVIDERS_SECTION); + return providers?.[provider]?.oauth; + } + + private abortExisting(provider: string): void { + const existing = this.flows.get(provider); + if (existing !== undefined && existing.status === 'pending') { + existing.controller.abort(); + this.setTerminal(existing, 'cancelled'); + } + } + + private invalidateFlows(): void { + for (const state of this.flows.values()) { + if (state.status === 'pending') { + state.controller.abort(); + } + if (state.gcTimer !== undefined) { + clearTimeout(state.gcTimer); + } + } + this.flows.clear(); + } + + private handleSuccess(state: FlowState): void { + if (state.status !== 'pending') return; + this.setTerminal(state, 'authenticated'); + } + + private handleFailure(state: FlowState, err: unknown): void { + if (state.status !== 'pending') return; + state.errorMessage = err instanceof Error ? err.message : String(err); + this.setTerminal(state, classifyFailure(err)); + } + + private setTerminal(state: FlowState, status: OAuthFlowStatus): void { + state.status = status; + state.resolvedAt = new Date().toISOString(); + const timer = setTimeout(() => { + if (this.flows.get(state.provider) === state) { + this.flows.delete(state.provider); + } + }, TERMINAL_RETENTION_MS); + timer.unref(); + state.gcTimer = timer; + } + + private toFlowStart(state: FlowState, device: DeviceAuthorization): OAuthFlowStart { + const expiresIn = device.expiresIn ?? DEFAULT_DEVICE_EXPIRES_IN_SEC; + return { + flow_id: state.flowId, + provider: state.provider, + verification_uri: device.verificationUri, + verification_uri_complete: device.verificationUriComplete, + user_code: device.userCode, + expires_in: expiresIn, + interval: device.interval, + status: 'pending', + expires_at: new Date(state.expiresAt).toISOString(), + }; + } + + private toSnapshot(state: FlowState, device: DeviceAuthorization): OAuthFlowSnapshot { + return { + ...this.toFlowStart(state, device), + status: state.status, + resolved_at: state.resolvedAt, + error_message: state.errorMessage, + }; + } } export class AuthSummaryService implements IAuthSummaryService { declare readonly _serviceBrand: undefined; constructor( - @IConfigService _config: IConfigService, - @ITelemetryService _telemetry: ITelemetryService, - private readonly oauth?: OAuthService, + @IConfigService private readonly config: IConfigService, + @IOAuthService private readonly oauth: IOAuthService, ) {} - summarize(): Promise { - if (this.oauth === undefined) return Promise.resolve([]); - return this.oauth.status().then((s) => [s]); + async summarize(): Promise { + const providers = this.config.get(PROVIDERS_SECTION) ?? {}; + const statuses: AuthStatus[] = []; + for (const [name, providerConfig] of Object.entries(providers)) { + if (providerConfig.oauth !== undefined) { + statuses.push(await this.oauth.status(name)); + } + } + return statuses; } + + async ensureReady(provider = KIMI_CODE_PROVIDER_NAME): Promise { + const status = await this.oauth.status(provider); + if (!status.loggedIn) { + throw new KimiError( + ErrorCodes.AUTH_LOGIN_REQUIRED, + `OAuth provider "${provider}" requires login before it can be used.`, + ); + } + } +} + +function classifyFailure(err: unknown): OAuthFlowStatus { + if (err instanceof DeviceCodeTimeoutError) return 'expired'; + if (err instanceof OAuthError) { + return err.message.toLowerCase().includes('aborted') ? 'cancelled' : 'denied'; + } + return 'denied'; } registerScopedService(LifecycleScope.Core, IOAuthService, OAuthService, InstantiationType.Delayed, 'auth'); diff --git a/packages/agent-core-v2/src/auth/index.ts b/packages/agent-core-v2/src/auth/index.ts index 7e0989751..b031e8ed4 100644 --- a/packages/agent-core-v2/src/auth/index.ts +++ b/packages/agent-core-v2/src/auth/index.ts @@ -6,3 +6,4 @@ export * from './auth'; export * from './authService'; +export * from './oauthSchemas'; diff --git a/packages/agent-core-v2/src/auth/oauthSchemas.ts b/packages/agent-core-v2/src/auth/oauthSchemas.ts new file mode 100644 index 000000000..c70daf07f --- /dev/null +++ b/packages/agent-core-v2/src/auth/oauthSchemas.ts @@ -0,0 +1,52 @@ +/** + * `auth` domain (cross-cutting) — config schemas for OAuth providers. + * + * Owns the `providers` config-section schema and the `OAuthRef` / + * `ProviderConfig` models consumed by `OAuthService` (and, later, by the + * `kosong` provider manager). Field names and types mirror + * `packages/agent-core/src/config/schema.ts` so the same `config.toml` stays + * compatible across the two engines; the snake_case TOML mapping is handled by + * the `config` persistence layer, not here. + */ + +import { z } from 'zod'; + +export const ProviderTypeSchema = z.enum([ + 'anthropic', + 'openai', + 'kimi', + 'google-genai', + 'openai_responses', + 'vertexai', +]); + +export type ProviderType = z.infer; + +export const OAuthRefSchema = z.object({ + storage: z.enum(['file', 'keyring']), + key: z.string().min(1), + oauthHost: z.string().min(1).optional(), +}); + +export type OAuthRef = z.infer; + +const StringRecordSchema = z.record(z.string(), z.string()); + +export const ProviderConfigSchema = z.object({ + type: ProviderTypeSchema, + apiKey: z.string().optional(), + baseUrl: z.string().optional(), + defaultModel: z.string().optional(), + oauth: OAuthRefSchema.optional(), + env: StringRecordSchema.optional(), + customHeaders: StringRecordSchema.optional(), + source: z.record(z.string(), z.unknown()).optional(), +}); + +export type ProviderConfig = z.infer; + +export const PROVIDERS_SECTION = 'providers'; + +export const ProvidersSectionSchema = z.record(z.string(), ProviderConfigSchema); + +export type ProvidersSection = z.infer; diff --git a/packages/agent-core-v2/src/config/config.ts b/packages/agent-core-v2/src/config/config.ts index b497c29c9..a0a6934f9 100644 --- a/packages/agent-core-v2/src/config/config.ts +++ b/packages/agent-core-v2/src/config/config.ts @@ -1,55 +1,107 @@ /** - * `config` domain (L2) — configuration registry, service, and per-agent view. + * `config` domain (L2) — configuration registry, global config service, and + * session-level runtime config service. * - * Defines the config service identifiers and the `ConfigSection` / - * `ConfigChangedEvent` models: the `IConfigRegistry` for section schemas, the - * `IConfigService` used to read and mutate config, and the per-agent - * `IAgentConfigService` view. The registry and service are Core-scoped; the - * agent view is Agent-scoped. + * Defines the config service identifiers and section models: the + * `IConfigRegistry` for section schemas, the Core-scoped `IConfigService` used + * to read and mutate global config, and the Session-scoped + * `ISessionConfigService` for the active session's runtime config. */ import type { Event } from '#/_base/event'; import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; -export interface ConfigSection { +export interface ConfigSchema { + parse(value: unknown): T; +} + +export type ConfigMerge = (base: T | undefined, patch: unknown) => T; + +export interface ConfigSection { readonly domain: string; - readonly schema: unknown; + readonly schema?: ConfigSchema; + readonly defaultValue?: T; + readonly merge: ConfigMerge; +} + +export interface RegisterSectionOptions { + readonly defaultValue?: T; + readonly merge?: ConfigMerge; } export interface IConfigRegistry { readonly _serviceBrand: undefined; - registerSection(domain: string, schema: unknown): void; + + registerSection(domain: string, schema: ConfigSchema, options?: RegisterSectionOptions): void; getSection(domain: string): ConfigSection | undefined; - merge(base: unknown, patch: unknown): unknown; + listSections(): readonly ConfigSection[]; + validate(domain: string, value: unknown): T; + merge(domain: string, base: T | undefined, patch: unknown): T; + defaultValue(domain: string): T | undefined; } export const IConfigRegistry: ServiceIdentifier = createDecorator('configRegistry'); +export type ConfigChangeSource = 'load' | 'reload' | 'set'; + export interface ConfigChangedEvent { readonly domain: string; + readonly source: ConfigChangeSource; } +export interface ConfigDiagnostic { + readonly domain?: string; + readonly severity: 'warning' | 'error'; + readonly message: string; +} + +export type ResolvedConfig = Record; + export interface IConfigService { readonly _serviceBrand: undefined; + readonly ready: Promise; readonly onDidChange: Event; + get(domain: string): T; + getAll(): ResolvedConfig; set(domain: string, patch: unknown): Promise; + reload(): Promise; + diagnostics(): readonly ConfigDiagnostic[]; } export const IConfigService: ServiceIdentifier = createDecorator('configService'); -export interface IAgentConfigService { +export interface SessionConfigSection { + readonly modelAlias?: string; + readonly thinkingLevel?: string; + readonly systemPrompt?: string; + readonly provider?: string; +} + +export type SessionConfigPatch = Partial< + Pick +>; + +export interface SessionConfigChangedEvent { + readonly changed: readonly (keyof SessionConfigSection)[]; +} + +export interface ISessionConfigService { readonly _serviceBrand: undefined; + readonly ready: Promise; + readonly onDidChange: Event; + readonly modelAlias: string | undefined; readonly thinkingLevel: string | undefined; readonly systemPrompt: string | undefined; readonly provider: string | undefined; - readonly cwd: string; + + update(patch: SessionConfigPatch): Promise; setModel(alias: string): Promise; setThinking(level: string): Promise; } -export const IAgentConfigService: ServiceIdentifier = - createDecorator('agentConfigService'); +export const ISessionConfigService: ServiceIdentifier = + createDecorator('sessionConfigService'); diff --git a/packages/agent-core-v2/src/config/configPure.ts b/packages/agent-core-v2/src/config/configPure.ts new file mode 100644 index 000000000..bcff7b6cc --- /dev/null +++ b/packages/agent-core-v2/src/config/configPure.ts @@ -0,0 +1,38 @@ +/** + * `config` domain (L2) — pure helper functions for config values. + * + * Provides side-effect-free helpers used by config services, including plain + * object detection, deep merge, undefined stripping, and error formatting. + */ + +export function isPlainObject(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +export function deepMerge(base: T | undefined, patch: unknown): T { + if (!isPlainObject(base) || !isPlainObject(patch)) { + return (patch ?? base) as T; + } + const out: Record = { ...base }; + for (const key of Object.keys(patch)) { + const pv = patch[key]; + const bv = out[key]; + out[key] = isPlainObject(bv) && isPlainObject(pv) ? deepMerge(bv, pv) : pv; + } + return out as T; +} + +export function omitUndefined>(value: T): Partial { + const out: Partial = {}; + for (const key of Object.keys(value)) { + const v = value[key]; + if (v !== undefined) { + out[key as keyof T] = v as T[keyof T]; + } + } + return out; +} + +export function describeUnknownError(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} diff --git a/packages/agent-core-v2/src/config/configService.ts b/packages/agent-core-v2/src/config/configService.ts index 0b07a6e18..92c56eda3 100644 --- a/packages/agent-core-v2/src/config/configService.ts +++ b/packages/agent-core-v2/src/config/configService.ts @@ -1,65 +1,78 @@ /** - * `config` domain (L2) — `IConfigRegistry`, `IConfigService`, and - * `IAgentConfigService` implementations. + * `config` domain (L2) — `IConfigRegistry` and `IConfigService` implementations. * - * Owns the in-memory config store, the section registry, and the per-agent - * config view; reads the environment through `environment`, resolves the agent - * cwd through `kaos`, records through `records`, and logs through `log`. Bound - * at Core (registry and service) and Agent (agent view) scopes. + * Owns the section registry and the global config file state; reads config + * paths through `environment` and logs through `log`. Bound at Core scope. */ +import { existsSync, readFileSync } from 'node:fs'; +import { mkdir } from 'node:fs/promises'; +import { dirname } from 'pathe'; + +import { parse as parseToml, stringify as stringifyToml } from 'smol-toml'; + import { Disposable } from '#/_base/di/lifecycle'; import { Emitter, type Event } from '#/_base/event'; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { IEnvironmentService } from '#/environment'; -import { IKaosService } from '#/kaos'; -import { ILogService } from '#/log'; -import { IAgentRecords } from '#/records'; +import { atomicWrite } from '#/_base/utils/fs'; +import { IEnvironmentService } from '#/environment/environment'; +import { ILogService } from '#/log/log'; import { type ConfigChangedEvent, + type ConfigDiagnostic, + type ConfigMerge, + type ConfigSchema, type ConfigSection, - IAgentConfigService, + type ConfigChangeSource, + type RegisterSectionOptions, + type ResolvedConfig, IConfigRegistry, IConfigService, } from './config'; - -function isPlainObject(value: unknown): value is Record { - return typeof value === 'object' && value !== null && !Array.isArray(value); -} - -function deepMerge(base: T, patch: unknown): T { - if (!isPlainObject(base) || !isPlainObject(patch)) { - return (patch ?? base) as T; - } - const out: Record = { ...base }; - for (const key of Object.keys(patch)) { - const pv = patch[key]; - const bv = out[key]; - out[key] = isPlainObject(bv) && isPlainObject(pv) ? deepMerge(bv, pv) : pv; - } - return out as T; -} +import { deepMerge, describeUnknownError, isPlainObject } from './configPure'; export class ConfigRegistry implements IConfigRegistry { declare readonly _serviceBrand: undefined; - private readonly sections = new Map(); + private readonly sections = new Map(); - registerSection(domain: string, schema: unknown): void { + registerSection( + domain: string, + schema: ConfigSchema, + options: RegisterSectionOptions = {}, + ): void { if (this.sections.has(domain)) { throw new Error(`ConfigRegistry: section '${domain}' is already registered`); } - this.sections.set(domain, schema); + this.sections.set(domain, { + domain, + schema: schema as ConfigSchema, + defaultValue: options.defaultValue, + merge: (options.merge ?? deepMerge) as ConfigMerge, + }); } getSection(domain: string): ConfigSection | undefined { - const schema = this.sections.get(domain); - return schema === undefined ? undefined : { domain, schema }; + return this.sections.get(domain); } - merge(base: unknown, patch: unknown): unknown { - return deepMerge(base, patch); + listSections(): readonly ConfigSection[] { + return [...this.sections.values()]; + } + + validate(domain: string, value: unknown): T { + const schema = this.sections.get(domain)?.schema; + return (schema === undefined ? value : schema.parse(value)) as T; + } + + merge(domain: string, base: T | undefined, patch: unknown): T { + const merge = this.sections.get(domain)?.merge ?? deepMerge; + return merge(base, patch) as T; + } + + defaultValue(domain: string): T | undefined { + return this.sections.get(domain)?.defaultValue as T | undefined; } } @@ -67,83 +80,115 @@ export class ConfigService extends Disposable implements IConfigService { declare readonly _serviceBrand: undefined; private readonly _onDidChange = this._register(new Emitter()); readonly onDidChange: Event = this._onDidChange.event; - private readonly root = new Map(); + readonly ready = Promise.resolve(); + + private raw: ResolvedConfig = {}; + private effective: ResolvedConfig = {}; + private readonly diagnosticsList: ConfigDiagnostic[] = []; constructor( - @IConfigRegistry _registry: IConfigRegistry, - @IEnvironmentService _env: IEnvironmentService, - @ILogService _log: ILogService, + @IConfigRegistry private readonly registry: IConfigRegistry, + @IEnvironmentService private readonly env: IEnvironmentService, + @ILogService private readonly log: ILogService, ) { super(); + this.loadSync('load'); } get(domain: string): T { - return this.root.get(domain) as T; + return this.effective[domain] as T; } - set(domain: string, patch: unknown): Promise { - const current = this.root.get(domain); - const next = deepMerge(current ?? {}, patch); - this.root.set(domain, next); - this._onDidChange.fire({ domain }); + getAll(): ResolvedConfig { + return { ...this.effective }; + } + + diagnostics(): readonly ConfigDiagnostic[] { + return [...this.diagnosticsList]; + } + + async set(domain: string, patch: unknown): Promise { + const currentRaw = this.raw; + const base = currentRaw[domain]; + const next = this.registry.merge(domain, base, patch); + const validated = this.registry.validate(domain, next); + const nextRaw: ResolvedConfig = { + ...currentRaw, + [domain]: validated, + }; + + await mkdir(dirname(this.env.configPath), { recursive: true, mode: 0o700 }); + await atomicWrite(this.env.configPath, `${stringifyToml(nextRaw)}\n`); + this.applyRaw(nextRaw, 'set', [domain]); + } + + reload(): Promise { + this.loadSync('reload'); return Promise.resolve(); } -} -interface AgentSection { - readonly modelAlias?: string; - readonly thinkingLevel?: string; - readonly systemPrompt?: string; - readonly provider?: string; -} - -export class AgentConfigService implements IAgentConfigService { - declare readonly _serviceBrand: undefined; - private modelAliasValue: string | undefined; - private thinkingLevelValue: string | undefined; - private systemPromptValue: string | undefined; - private providerValue: string | undefined; - private readonly cwdValue: string; - - constructor( - @IConfigService config: IConfigService, - @IAgentRecords _records: IAgentRecords, - @IKaosService agentKaos: IKaosService, - ) { - const section = config.get('agent'); - this.modelAliasValue = section?.modelAlias; - this.thinkingLevelValue = section?.thinkingLevel; - this.systemPromptValue = section?.systemPrompt; - this.providerValue = section?.provider; - this.cwdValue = agentKaos.cwd; + private loadSync(source: ConfigChangeSource): void { + this.diagnosticsList.length = 0; + let raw: ResolvedConfig = {}; + try { + raw = this.readRawFileSync(); + } catch (error) { + this.diagnosticsList.push({ + severity: 'error', + message: describeUnknownError(error), + }); + this.log.warn('config load failed', { error: describeUnknownError(error) }); + } + this.applyRaw(raw, source); } - get modelAlias(): string | undefined { - return this.modelAliasValue; - } - get thinkingLevel(): string | undefined { - return this.thinkingLevelValue; - } - get systemPrompt(): string | undefined { - return this.systemPromptValue; - } - get provider(): string | undefined { - return this.providerValue; - } - get cwd(): string { - return this.cwdValue; + private readRawFileSync(): ResolvedConfig { + if (!existsSync(this.env.configPath)) { + return {}; + } + const text = readFileSync(this.env.configPath, 'utf-8'); + if (text.trim().length === 0) { + return {}; + } + try { + const parsed = parseToml(text); + return isPlainObject(parsed) ? parsed : {}; + } catch (error) { + throw new Error(`Failed to parse ${this.env.configPath}: ${describeUnknownError(error)}`); + } } - setModel(alias: string): Promise { - this.modelAliasValue = alias; - return Promise.resolve(); + private applyRaw(raw: ResolvedConfig, source: ConfigChangeSource, domains?: readonly string[]): void { + const previous = this.raw; + this.raw = raw; + this.effective = this.buildEffective(raw); + const changedDomains = domains ?? [...new Set([...Object.keys(previous), ...Object.keys(raw)])]; + for (const domain of changedDomains) { + this._onDidChange.fire({ domain, source }); + } } - setThinking(level: string): Promise { - this.thinkingLevelValue = level; - return Promise.resolve(); + + private buildEffective(raw: ResolvedConfig): ResolvedConfig { + const effective: ResolvedConfig = {}; + for (const [domain, value] of Object.entries(raw)) { + try { + effective[domain] = this.registry.validate(domain, value); + } catch (error) { + this.diagnosticsList.push({ + domain, + severity: 'warning', + message: `Ignored invalid config section '${domain}': ${describeUnknownError(error)}`, + }); + } + } + for (const section of this.registry.listSections()) { + if (effective[section.domain] === undefined && section.defaultValue !== undefined) { + effective[section.domain] = section.defaultValue; + } + } + return effective; } } registerScopedService(LifecycleScope.Core, IConfigRegistry, ConfigRegistry, InstantiationType.Delayed, 'config'); registerScopedService(LifecycleScope.Core, IConfigService, ConfigService, InstantiationType.Delayed, 'config'); -registerScopedService(LifecycleScope.Agent, IAgentConfigService, AgentConfigService, InstantiationType.Delayed, 'config'); diff --git a/packages/agent-core-v2/src/config/index.ts b/packages/agent-core-v2/src/config/index.ts index 2aeac6632..2282521c7 100644 --- a/packages/agent-core-v2/src/config/index.ts +++ b/packages/agent-core-v2/src/config/index.ts @@ -6,3 +6,4 @@ export * from './config'; export * from './configService'; +export * from './sessionConfigService'; diff --git a/packages/agent-core-v2/src/config/sessionConfigService.ts b/packages/agent-core-v2/src/config/sessionConfigService.ts new file mode 100644 index 000000000..c1632ddf9 --- /dev/null +++ b/packages/agent-core-v2/src/config/sessionConfigService.ts @@ -0,0 +1,114 @@ +/** + * `config` domain (L2) — `ISessionConfigService` implementation. + * + * Owns the active session's runtime config overrides; reads global defaults + * through `config`, persists session-level overrides through `records`, and logs + * through `log`. Bound at Session scope. + */ + +import { Disposable } from '#/_base/di/lifecycle'; +import { Emitter, type Event } from '#/_base/event'; +import { InstantiationType } from '#/_base/di/extensions'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { ILogService } from '#/log/log'; +import { ISessionMetaStore } from '#/records/records'; + +import { + type SessionConfigChangedEvent, + type SessionConfigPatch, + type SessionConfigSection, + IConfigService, + ISessionConfigService, +} from './config'; +import { describeUnknownError, omitUndefined } from './configPure'; + +export class SessionConfigService extends Disposable implements ISessionConfigService { + declare readonly _serviceBrand: undefined; + private readonly _onDidChange = this._register(new Emitter()); + readonly onDidChange: Event = this._onDidChange.event; + readonly ready: Promise; + + private modelAliasValue: string | undefined; + private thinkingLevelValue: string | undefined; + private systemPromptValue: string | undefined; + private providerValue: string | undefined; + + constructor( + @IConfigService config: IConfigService, + @ISessionMetaStore private readonly meta: ISessionMetaStore, + @ILogService private readonly log: ILogService, + ) { + super(); + const section = config.get('session') ?? {}; + this.apply(section, false); + this.ready = this.restore(); + } + + get modelAlias(): string | undefined { + return this.modelAliasValue; + } + get thinkingLevel(): string | undefined { + return this.thinkingLevelValue; + } + get systemPrompt(): string | undefined { + return this.systemPromptValue; + } + get provider(): string | undefined { + return this.providerValue; + } + + async update(patch: SessionConfigPatch): Promise { + const clean = omitUndefined(patch as Record) as SessionConfigPatch; + if (Object.keys(clean).length === 0) return; + await this.meta.write(clean); + this.apply(clean, true); + } + + setModel(alias: string): Promise { + return this.update({ modelAlias: alias }); + } + + setThinking(level: string): Promise { + return this.update({ thinkingLevel: level }); + } + + private async restore(): Promise { + try { + const stored = await this.meta.read(); + this.apply(stored as Partial, false); + } catch (error) { + this.log.warn('session config restore failed', { error: describeUnknownError(error) }); + } + } + + private apply(patch: Partial, emit: boolean): void { + const changed: (keyof SessionConfigSection)[] = []; + if (patch.modelAlias !== undefined && patch.modelAlias !== this.modelAliasValue) { + this.modelAliasValue = patch.modelAlias; + changed.push('modelAlias'); + } + if (patch.thinkingLevel !== undefined && patch.thinkingLevel !== this.thinkingLevelValue) { + this.thinkingLevelValue = patch.thinkingLevel; + changed.push('thinkingLevel'); + } + if (patch.systemPrompt !== undefined && patch.systemPrompt !== this.systemPromptValue) { + this.systemPromptValue = patch.systemPrompt; + changed.push('systemPrompt'); + } + if (patch.provider !== undefined && patch.provider !== this.providerValue) { + this.providerValue = patch.provider; + changed.push('provider'); + } + if (emit && changed.length > 0) { + this._onDidChange.fire({ changed }); + } + } +} + +registerScopedService( + LifecycleScope.Session, + ISessionConfigService, + SessionConfigService, + InstantiationType.Delayed, + 'config', +); diff --git a/packages/agent-core-v2/src/gateway/gateway.ts b/packages/agent-core-v2/src/gateway/gateway.ts index 924887236..424ffc16f 100644 --- a/packages/agent-core-v2/src/gateway/gateway.ts +++ b/packages/agent-core-v2/src/gateway/gateway.ts @@ -31,6 +31,8 @@ export interface IRestGateway { steer(sessionId: string, agentId: string, content: string): Promise; cancel(sessionId: string, agentId: string, reason?: string): Promise; getStatus(sessionId: string): Promise; + flushLogs(sessionId: string): Promise; + flushGlobalLogs(): Promise; } export const IRestGateway: ServiceIdentifier = diff --git a/packages/agent-core-v2/src/gateway/gatewayService.ts b/packages/agent-core-v2/src/gateway/gatewayService.ts index 1f89d39be..83f11ef67 100644 --- a/packages/agent-core-v2/src/gateway/gatewayService.ts +++ b/packages/agent-core-v2/src/gateway/gatewayService.ts @@ -3,8 +3,8 @@ * `IWSBroadcastService` implementation. * * Owns the session scope registry and the REST/WS entry points; resolves agents - * through `agent-lifecycle`, drives turns through `turn`, and subscribes to - * broadcasts through `event`. Bound at Core scope. + * through `agent-lifecycle`, drives turns through `turn`, flushes logs through + * `log`, and subscribes to broadcasts through `event`. Bound at Core scope. */ import { Disposable } from '#/_base/di/lifecycle'; @@ -17,8 +17,9 @@ import { } from '#/_base/di/scope'; import { IInstantiationService } from '#/_base/di/instantiation'; import { IAgentLifecycleService } from '#/agent-lifecycle/agentLifecycle'; -import { IEventService } from '#/event'; -import { ITurnService } from '#/turn'; +import { IEventService } from '#/event/event'; +import { ILogService, ISessionLogService } from '#/log/log'; +import { ITurnService } from '#/turn/turn'; import { type CreateSessionOptions, @@ -57,7 +58,10 @@ export class ScopeRegistry implements IScopeRegistry { export class RestGateway implements IRestGateway { declare readonly _serviceBrand: undefined; - constructor(@IScopeRegistry private readonly scopes: IScopeRegistry) {} + constructor( + @IScopeRegistry private readonly scopes: IScopeRegistry, + @ILogService private readonly log: ILogService, + ) {} private turn(sessionId: string, agentId: string): ITurnService { const session = this.scopes.get(sessionId); @@ -82,6 +86,16 @@ export class RestGateway implements IRestGateway { getStatus(sessionId: string): Promise { return Promise.resolve(this.scopes.get(sessionId) !== undefined); } + + async flushLogs(sessionId: string): Promise { + const session = this.scopes.get(sessionId); + if (session === undefined) return; + await session.accessor.get(ISessionLogService).flush(); + } + + flushGlobalLogs(): Promise { + return this.log.flush(); + } } export class WSGateway implements IWSGateway { diff --git a/packages/agent-core-v2/src/kaos/kaos.ts b/packages/agent-core-v2/src/kaos/kaos.ts index 9e1657e23..e7d3aba40 100644 --- a/packages/agent-core-v2/src/kaos/kaos.ts +++ b/packages/agent-core-v2/src/kaos/kaos.ts @@ -5,6 +5,8 @@ import { createDecorator } from '#/_base/di/instantiation'; export interface IKaosService { readonly _serviceBrand: undefined; readonly kaos: Kaos | undefined; + readonly cwd: string; + chdir(cwd: string): Promise; } // eslint-disable-next-line @typescript-eslint/no-redeclare diff --git a/packages/agent-core-v2/src/kaos/kaosService.ts b/packages/agent-core-v2/src/kaos/kaosService.ts index 918158c84..6064b6d2f 100644 --- a/packages/agent-core-v2/src/kaos/kaosService.ts +++ b/packages/agent-core-v2/src/kaos/kaosService.ts @@ -16,6 +16,18 @@ export class KaosService implements IKaosService { get kaos(): Kaos | undefined { return this.options.kaos; } + + get cwd(): string { + const kaos = this.options.kaos; + if (kaos === undefined) { + throw new Error('KaosService.cwd accessed before kaos was provided'); + } + return kaos.getcwd(); + } + + chdir(): Promise { + return Promise.reject(new Error('KaosService.chdir is not supported')); + } } registerScopedService( diff --git a/packages/agent-core-v2/src/kosong/kosongService.ts b/packages/agent-core-v2/src/kosong/kosongService.ts index c1e28c5f5..36f3001e4 100644 --- a/packages/agent-core-v2/src/kosong/kosongService.ts +++ b/packages/agent-core-v2/src/kosong/kosongService.ts @@ -9,8 +9,8 @@ import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { IAgentConfigService, IConfigService } from '#/config'; -import { IEnvironmentService } from '#/environment'; +import { IConfigService, ISessionConfigService } from '#/config/config'; +import { IEnvironmentService } from '#/environment/environment'; import { type GenerateArgs, @@ -86,14 +86,14 @@ export class LLMService implements ILLMService { constructor( @IProviderManager private readonly providers: IProviderManager, - @IAgentConfigService private readonly agentConfig: IAgentConfigService, + @ISessionConfigService private readonly sessionConfig: ISessionConfigService, ) {} // eslint-disable-next-line require-yield -- TODO stub: yields the kosong stream once wired. async *generate(_args: GenerateArgs): AsyncIterable { const resolved = await this.providers.resolve( - this.agentConfig.provider, - this.agentConfig.modelAlias, + this.sessionConfig.provider, + this.sessionConfig.modelAlias, ); throw new Error(`TODO: LLMService.generate (${resolved.providerId}/${resolved.modelId})`); } diff --git a/packages/agent-core-v2/src/permission/permission.ts b/packages/agent-core-v2/src/permission/permission.ts index 789174508..d4546abf2 100644 --- a/packages/agent-core-v2/src/permission/permission.ts +++ b/packages/agent-core-v2/src/permission/permission.ts @@ -2,7 +2,7 @@ import type { PermissionData, PermissionMode, } from '../../../agent/permission'; -import { createDecorator } from "#/_base/di"; +import { createDecorator } from "#/_base/di/instantiation"; import type { AuthorizeToolExecutionResult, ResolvedToolExecutionHookContext, diff --git a/packages/agent-core-v2/src/telemetry/cloudSink.ts b/packages/agent-core-v2/src/telemetry/cloudSink.ts new file mode 100644 index 000000000..b543e1d97 --- /dev/null +++ b/packages/agent-core-v2/src/telemetry/cloudSink.ts @@ -0,0 +1,169 @@ +/** + * `telemetry` domain (L1) — `CloudSink`, a `TelemetryClient` that batches + * events, enriches them with common context, and posts them to the telemetry + * endpoint through `CloudTransport`. Core-scoped; has no cross-domain + * collaborators and is independent of `@moonshot-ai/kimi-telemetry`. + */ + +import { randomUUID } from 'node:crypto'; +import { arch, platform, release } from 'node:os'; + +import type { TelemetryClient, TelemetryContextPatch, TelemetryProperties } from './telemetry'; +import { + type CloudContext, + type CloudPrimitive, + type CloudProperties, + CloudTransport, + type EnrichedCloudEvent, + isCloudPrimitive, +} from './cloudTransport'; + +export interface CloudSinkOptions { + readonly homeDir: string; + readonly deviceId: string; + readonly sessionId?: string; + readonly appName: string; + readonly version: string; + readonly uiMode?: string; + readonly model?: string; + readonly buildSha?: string; + readonly terminal?: string; + readonly locale?: string; + readonly getAccessToken?: () => string | null | Promise; + readonly endpoint?: string; + readonly flushThreshold?: number; + readonly flushIntervalMs?: number; + readonly fetchImpl?: typeof fetch; + readonly retryBackoffsMs?: readonly number[]; + readonly requestTimeoutMs?: number; + readonly sleep?: (ms: number, signal?: AbortSignal) => Promise; + readonly now?: () => number; + readonly env?: NodeJS.ProcessEnv; +} + +const DEFAULT_FLUSH_THRESHOLD = 50; +const DEFAULT_FLUSH_INTERVAL_MS = 30_000; + +export class CloudSink implements TelemetryClient { + private readonly transport: CloudTransport; + private readonly context: CloudContext; + private readonly flushThreshold: number; + private readonly flushIntervalMs: number; + private deviceId: string; + private sessionId: string | null; + private buffer: EnrichedCloudEvent[] = []; + private flushTimer: ReturnType | null = null; + + constructor(options: CloudSinkOptions) { + this.deviceId = options.deviceId; + this.sessionId = options.sessionId ?? null; + this.flushThreshold = options.flushThreshold ?? DEFAULT_FLUSH_THRESHOLD; + this.flushIntervalMs = options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS; + this.context = buildContext(options); + this.transport = new CloudTransport({ + homeDir: options.homeDir, + deviceId: options.deviceId, + endpoint: options.endpoint, + getAccessToken: options.getAccessToken, + fetchImpl: options.fetchImpl, + retryBackoffsMs: options.retryBackoffsMs, + requestTimeoutMs: options.requestTimeoutMs, + sleep: options.sleep, + now: options.now, + }); + } + + track(event: string, properties?: TelemetryProperties): void { + const enriched: EnrichedCloudEvent = { + event_id: randomUUID().replaceAll('-', ''), + device_id: this.deviceId, + session_id: this.sessionId, + event, + timestamp: Date.now() / 1000, + properties: sanitizeProperties(properties), + context: { ...this.context }, + }; + this.buffer.push(enriched); + if (this.buffer.length >= this.flushThreshold) { + void this.flush().catch(() => {}); + } + } + + setContext(patch: TelemetryContextPatch): void { + const deviceId = patch['deviceId']; + if (typeof deviceId === 'string') { + this.deviceId = deviceId; + } + const sessionId = patch['sessionId']; + if (typeof sessionId === 'string') { + this.sessionId = sessionId; + } + } + + async flush(): Promise { + if (this.buffer.length === 0) return; + const events = this.buffer; + this.buffer = []; + await this.transport.send(events); + } + + async shutdown(): Promise { + this.stopPeriodicFlush(); + await this.flush(); + } + + startPeriodicFlush(): void { + if (this.flushTimer !== null) return; + this.flushTimer = setInterval(() => { + void this.flush().catch(() => {}); + }, this.flushIntervalMs); + this.flushTimer.unref?.(); + } + + stopPeriodicFlush(): void { + if (this.flushTimer === null) return; + clearInterval(this.flushTimer); + this.flushTimer = null; + } + + async retryDiskEvents(): Promise { + await this.transport.retryDiskEvents(); + } +} + +function sanitizeProperties(input?: TelemetryProperties): CloudProperties { + const out: CloudProperties = {}; + if (input === undefined) return out; + for (const [key, value] of Object.entries(input)) { + if (isCloudPrimitive(value)) { + out[key] = value; + } + } + return out; +} + +function buildContext(options: CloudSinkOptions): CloudContext { + const env = options.env ?? process.env; + const context: CloudContext = { + app_name: options.appName, + version: options.version, + runtime: 'node', + platform: platform(), + arch: arch(), + node_version: process.versions.node, + os_version: release(), + ci: env['CI'] !== undefined, + locale: options.locale ?? env['LANG'] ?? '', + terminal: options.terminal ?? env['TERM_PROGRAM'] ?? '', + ui_mode: options.uiMode ?? 'shell', + }; + setPrimitive(context, 'model', options.model); + setPrimitive(context, 'build_sha', options.buildSha); + return context; +} + +function setPrimitive(target: CloudContext, key: string, value: CloudPrimitive): void { + if (value === undefined) return; + if (typeof value === 'string' && value.length === 0) return; + target[key] = value; +} diff --git a/packages/agent-core-v2/src/telemetry/cloudTransport.ts b/packages/agent-core-v2/src/telemetry/cloudTransport.ts new file mode 100644 index 000000000..d497b17af --- /dev/null +++ b/packages/agent-core-v2/src/telemetry/cloudTransport.ts @@ -0,0 +1,383 @@ +/** + * `telemetry` domain (L1) — `CloudTransport`, the HTTP transport behind + * `CloudSink`. Posts enriched events to the telemetry endpoint with Bearer + * auth, retry, and on-disk fallback for failed events. Core-scoped; has no + * cross-domain collaborators and is independent of `@moonshot-ai/kimi-telemetry`. + */ + +import { randomBytes } from 'node:crypto'; +import { + chmodSync, + mkdirSync, + readdirSync, + readFileSync, + statSync, + unlinkSync, + writeFileSync, +} from 'node:fs'; +import { join } from 'node:path'; + +export type CloudPrimitive = boolean | number | string | undefined | null; + +export type CloudProperties = Record; + +export type CloudContext = Record; + +export interface CloudEvent { + readonly event_id: string; + device_id: string | null; + session_id: string | null; + readonly event: string; + readonly timestamp: number; + readonly properties: CloudProperties; +} + +export interface EnrichedCloudEvent extends CloudEvent { + readonly context: CloudContext; +} + +export interface CloudPayload { + readonly user_id: string; + readonly events: readonly Record[]; +} + +export interface CloudTransportOptions { + readonly homeDir: string; + readonly deviceId: string; + readonly endpoint?: string; + readonly getAccessToken?: () => string | null | Promise; + readonly fetchImpl?: typeof fetch; + readonly retryBackoffsMs?: readonly number[]; + readonly requestTimeoutMs?: number; + readonly sleep?: (ms: number, signal?: AbortSignal) => Promise; + readonly now?: () => number; +} + +export const TELEMETRY_ENDPOINT = 'https://telemetry-logs.kimi.com/v1/event'; +export const SERVER_EVENT_PREFIX = 'kfc_'; +export const USER_ID_PREFIX = 'kfc_device_id_'; +export const DISK_EVENT_MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000; +export const RETRY_BACKOFFS_MS = [1_000, 4_000, 16_000] as const; + +const DEFAULT_REQUEST_TIMEOUT_MS = 10_000; + +export class CloudTransport { + private readonly homeDir: string; + private readonly deviceId: string; + private readonly endpoint: string; + private readonly getAccessToken: (() => string | null | Promise) | null; + private readonly fetchImpl: typeof fetch; + private readonly retryBackoffsMs: readonly number[]; + private readonly requestTimeoutMs: number; + private readonly sleepImpl: (ms: number, signal?: AbortSignal) => Promise; + private readonly now: () => number; + + constructor(options: CloudTransportOptions) { + this.homeDir = options.homeDir; + this.deviceId = options.deviceId; + this.endpoint = options.endpoint ?? TELEMETRY_ENDPOINT; + this.getAccessToken = options.getAccessToken ?? null; + this.fetchImpl = options.fetchImpl ?? globalThis.fetch.bind(globalThis); + this.retryBackoffsMs = options.retryBackoffsMs ?? RETRY_BACKOFFS_MS; + this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS; + this.sleepImpl = options.sleep ?? abortableSleep; + this.now = options.now ?? Date.now; + } + + async send(events: readonly EnrichedCloudEvent[], signal?: AbortSignal): Promise { + if (events.length === 0) return; + let savedToDisk = false; + const saveEventsToDisk = (): void => { + if (savedToDisk) return; + this.saveToDisk(events); + savedToDisk = true; + }; + if (signal?.aborted === true) { + saveEventsToDisk(); + throw abortError(); + } + + let payload: CloudPayload; + try { + payload = buildPayload(events, this.deviceId); + } catch { + return; + } + + try { + for (let attempt = 0; attempt <= this.retryBackoffsMs.length; attempt++) { + try { + await this.sendHttp(payload, signal); + return; + } catch (error) { + if (isSignalAborted(signal) || isAbortError(error)) { + saveEventsToDisk(); + throw error; + } + if (!(error instanceof TransientCloudError)) { + break; + } + const backoff = this.retryBackoffsMs[attempt]; + if (backoff === undefined) break; + await this.sleepImpl(backoff, signal); + } + } + } catch (error) { + if (isSignalAborted(signal) || isAbortError(error)) { + saveEventsToDisk(); + throw error; + } + } + + saveEventsToDisk(); + } + + saveToDisk(events: readonly EnrichedCloudEvent[]): void { + if (events.length === 0) return; + const path = join(this.telemetryDir(), `failed_${randomBytes(6).toString('hex')}.jsonl`); + const text = events.map((event) => JSON.stringify(event)).join('\n') + '\n'; + writeFileSync(path, text, { encoding: 'utf-8', mode: 0o600, flag: 'wx' }); + try { + chmodSync(path, 0o600); + } catch { + // best effort on platforms that do not support chmod. + } + } + + async retryDiskEvents(): Promise { + let entries: string[]; + try { + entries = readdirSync(this.telemetryDir()); + } catch { + return; + } + + const now = this.now(); + for (const entry of entries) { + if (!entry.startsWith('failed_') || !entry.endsWith('.jsonl')) continue; + const path = join(this.telemetryDir(), entry); + try { + const stat = statSync(path); + if (now - stat.mtimeMs > DISK_EVENT_MAX_AGE_MS) { + unlinkSync(path); + continue; + } + } catch { + continue; + } + + let events: EnrichedCloudEvent[]; + let payload: CloudPayload; + try { + events = readJsonl(path); + payload = buildPayload(events, this.deviceId); + } catch (error) { + if (error instanceof SyntaxError || error instanceof TypeError) { + try { + unlinkSync(path); + } catch { + // best effort cleanup. + } + } + continue; + } + + try { + await this.sendHttp(payload); + unlinkSync(path); + } catch (error) { + if (error instanceof TransientCloudError) continue; + } + } + } + + private async sendHttp(payload: CloudPayload, signal?: AbortSignal): Promise { + const token = this.getAccessToken === null ? null : await this.getAccessToken(); + const headers: Record = { + 'Content-Type': 'application/json', + }; + if (token !== null && token.length > 0) { + headers['Authorization'] = `Bearer ${token}`; + } + + const response = await this.post(payload, headers, signal); + if (response.status === 401 && headers['Authorization'] !== undefined) { + delete headers['Authorization']; + const retry = await this.post(payload, headers, signal); + handleStatus(retry.status); + return; + } + handleStatus(response.status); + } + + private async post( + payload: CloudPayload, + headers: Record, + signal?: AbortSignal, + ): Promise { + try { + return await fetchWithTimeout( + this.fetchImpl, + this.endpoint, + { + method: 'POST', + headers: { ...headers }, + body: JSON.stringify(payload), + }, + this.requestTimeoutMs, + signal, + ); + } catch (error) { + if (signal?.aborted === true || isAbortError(error)) throw error; + throw new TransientCloudError(String(error)); + } + } + + private telemetryDir(): string { + const path = join(this.homeDir, 'telemetry'); + mkdirSync(path, { recursive: true, mode: 0o700 }); + try { + chmodSync(path, 0o700); + } catch { + // best effort on platforms that do not support chmod. + } + return path; + } +} + +export class TransientCloudError extends Error { + override readonly name = 'TransientCloudError'; +} + +export function buildUserId(deviceId: string): string { + return USER_ID_PREFIX + deviceId; +} + +export function buildPayload( + events: readonly EnrichedCloudEvent[], + deviceId: string, +): CloudPayload { + return { + user_id: buildUserId(deviceId), + events: events.map((event) => flattenEvent(applyServerPrefix(event))), + }; +} + +export function applyServerPrefix(event: EnrichedCloudEvent): EnrichedCloudEvent { + const name: unknown = event.event; + if (typeof name !== 'string' || name.length === 0 || name.startsWith(SERVER_EVENT_PREFIX)) { + return event; + } + return { ...event, event: SERVER_EVENT_PREFIX + name }; +} + +export function flattenEvent(event: EnrichedCloudEvent): Record { + const out: Record = {}; + for (const [key, value] of Object.entries(event)) { + if (key === 'properties') { + flattenNested(out, 'property', value); + } else if (key === 'context') { + flattenNested(out, 'context', value); + } else { + assertPrimitive(key, value); + out[key] = value; + } + } + return out; +} + +export function isCloudPrimitive(value: unknown): value is CloudPrimitive { + return ( + value === null || + value === undefined || + typeof value === 'boolean' || + typeof value === 'string' || + (typeof value === 'number' && Number.isFinite(value)) + ); +} + +function flattenNested(target: Record, prefix: string, value: unknown) { + if (value === null || typeof value !== 'object' || Array.isArray(value)) return; + for (const [key, nestedValue] of Object.entries(value)) { + assertPrimitive(`${prefix}.${key}`, nestedValue); + target[`${prefix}_${key}`] = nestedValue; + } +} + +function assertPrimitive(key: string, value: unknown): asserts value is CloudPrimitive { + if (isCloudPrimitive(value)) return; + throw new TypeError(`telemetry ${key} must be primitive`); +} + +function handleStatus(status: number): void { + if (status >= 500 || status === 429) { + throw new TransientCloudError(`HTTP ${String(status)}`); + } + if (status >= 400) { + return; + } +} + +function readJsonl(path: string): EnrichedCloudEvent[] { + const text = readFileSync(path, 'utf-8'); + const events: EnrichedCloudEvent[] = []; + for (const line of text.split('\n')) { + const trimmed = line.trim(); + if (trimmed.length === 0) continue; + events.push(JSON.parse(trimmed) as EnrichedCloudEvent); + } + return events; +} + +async function fetchWithTimeout( + fetchImpl: typeof fetch, + url: string, + init: RequestInit, + timeoutMs: number, + externalSignal?: AbortSignal, +): Promise { + const controller = new AbortController(); + const abortFromExternal = (): void => { + controller.abort(externalSignal?.reason); + }; + const timeout = setTimeout(() => { + controller.abort(new Error('telemetry request timed out')); + }, timeoutMs); + timeout.unref?.(); + if (externalSignal?.aborted === true) abortFromExternal(); + externalSignal?.addEventListener('abort', abortFromExternal, { once: true }); + try { + return await fetchImpl(url, { + ...init, + signal: controller.signal, + }); + } finally { + clearTimeout(timeout); + externalSignal?.removeEventListener('abort', abortFromExternal); + } +} + +function abortableSleep(ms: number, signal?: AbortSignal): Promise { + if (signal?.aborted === true) return Promise.reject(abortError()); + return new Promise((resolve, reject) => { + const timer = setTimeout(resolve, ms); + timer.unref?.(); + const onAbort = (): void => { + clearTimeout(timer); + reject(abortError()); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + }); +} + +function isAbortError(error: unknown): boolean { + return error instanceof Error && error.name === 'AbortError'; +} + +function isSignalAborted(signal?: AbortSignal): boolean { + return signal?.aborted === true; +} + +function abortError(): DOMException { + return new DOMException('The operation was aborted.', 'AbortError'); +} diff --git a/packages/agent-core-v2/src/telemetry/consoleSink.ts b/packages/agent-core-v2/src/telemetry/consoleSink.ts new file mode 100644 index 000000000..1ec835290 --- /dev/null +++ b/packages/agent-core-v2/src/telemetry/consoleSink.ts @@ -0,0 +1,45 @@ +/** + * `telemetry` domain (L1) — `ConsoleSink`, a `TelemetryClient` that echoes + * events to a log function for development and debugging. Core-scoped; has no + * cross-domain collaborators. + */ + +import type { TelemetryClient, TelemetryProperties } from './telemetry'; + +export interface ConsoleSinkOptions { + readonly prefix?: string; + readonly pretty?: boolean; + readonly log?: (message: string) => void; +} + +const DEFAULT_PREFIX = '[telemetry]'; + +export class ConsoleSink implements TelemetryClient { + private readonly prefix: string; + private readonly pretty: boolean; + private readonly log: (message: string) => void; + + constructor(options: ConsoleSinkOptions = {}) { + this.prefix = options.prefix ?? DEFAULT_PREFIX; + this.pretty = options.pretty ?? false; + this.log = options.log ?? defaultLog; + } + + track(event: string, properties?: TelemetryProperties): void { + const payload = + properties === undefined ? '' : ` ${stringifyProperties(properties, this.pretty)}`; + this.log(`${this.prefix} ${event}${payload}`); + } +} + +function stringifyProperties(properties: TelemetryProperties, pretty: boolean): string { + if (pretty) { + return JSON.stringify(properties, null, 2); + } + return JSON.stringify(properties); +} + +function defaultLog(message: string): void { + // eslint-disable-next-line no-console + console.log(message); +} diff --git a/packages/agent-core-v2/src/telemetry/index.ts b/packages/agent-core-v2/src/telemetry/index.ts index 39e02917d..be9704798 100644 --- a/packages/agent-core-v2/src/telemetry/index.ts +++ b/packages/agent-core-v2/src/telemetry/index.ts @@ -1,8 +1,11 @@ /** - * `telemetry` domain barrel — re-exports the `telemetry` contract and its - * scoped service (`telemetryService`). Importing this barrel registers the - * `ITelemetryService` binding into the scope registry. + * `telemetry` domain barrel — re-exports the `telemetry` contract, its scoped + * service (`telemetryService`), and the bundled sinks (`ConsoleSink`, + * `CloudSink`). Importing this barrel registers the `ITelemetryService` + * binding into the scope registry. */ export * from './telemetry'; export * from './telemetryService'; +export * from './consoleSink'; +export * from './cloudSink'; diff --git a/packages/agent-core-v2/src/telemetry/telemetry.ts b/packages/agent-core-v2/src/telemetry/telemetry.ts index 3e6f1f16a..7a673ac03 100644 --- a/packages/agent-core-v2/src/telemetry/telemetry.ts +++ b/packages/agent-core-v2/src/telemetry/telemetry.ts @@ -1,4 +1,16 @@ +/** + * `telemetry` domain (L1) — `ITelemetryService` contract and sink types. + * + * Layer-1 root service: merges bound context into tracked events and fans + * them out to one or more `TelemetryClient` sinks. Core-scoped — stateless + * beyond its sink set and bound context; enrichment, batching, and transport + * are owned by the sinks, not by this layer. Defines the `TelemetryClient` + * sink contract, the `ITelemetryService` facade, the service options, and the + * no-op sink. + */ + import { createDecorator } from '#/_base/di/instantiation'; +import type { IDisposable } from '#/_base/di/lifecycle'; export type TelemetryPropertyValue = unknown; @@ -10,10 +22,13 @@ export interface TelemetryClient { track(event: string, properties?: TelemetryProperties): void; withContext?(patch: TelemetryContextPatch): TelemetryClient; setContext?(patch: TelemetryContextPatch): void; + flush?(): Promise | void; + shutdown?(): Promise | void; } export interface TelemetryServiceOptions { readonly client?: TelemetryClient; + readonly clients?: readonly TelemetryClient[]; readonly context?: TelemetryProperties; readonly sessionId?: string; readonly agentId?: string; @@ -21,15 +36,24 @@ export interface TelemetryServiceOptions { } export interface ITelemetryService { + readonly _serviceBrand: undefined; track(event: string, properties?: TelemetryProperties): void; withContext(patch: TelemetryContextPatch): ITelemetryService; setContext(patch: TelemetryContextPatch): void; + addSink(client: TelemetryClient): IDisposable; + removeSink(client: TelemetryClient): void; + setDelegate(client: TelemetryClient): void; + setEnabled(enabled: boolean): void; + flush(): Promise; + shutdown(): Promise; } export const noopTelemetryClient: TelemetryClient = { track: () => {}, withContext: () => noopTelemetryClient, setContext: () => {}, + flush: () => {}, + shutdown: () => {}, }; // eslint-disable-next-line @typescript-eslint/no-redeclare diff --git a/packages/agent-core-v2/src/telemetry/telemetryService.ts b/packages/agent-core-v2/src/telemetry/telemetryService.ts index 836ef672b..8d8609d26 100644 --- a/packages/agent-core-v2/src/telemetry/telemetryService.ts +++ b/packages/agent-core-v2/src/telemetry/telemetryService.ts @@ -1,3 +1,17 @@ +/** + * `telemetry` domain (L1) — `ITelemetryService` implementation. + * + * Merges bound context into each tracked event and fans it out to the + * registered `TelemetryClient` sinks; owns the sink set, the enabled flag, + * and the bound context, but no enrichment or transport of its own. Bound at + * Core scope; has no cross-domain collaborators. + */ + +import { InstantiationType } from '#/_base/di/extensions'; +import { type IDisposable, toDisposable } from '#/_base/di/lifecycle'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { onUnexpectedError } from '#/_base/errors/unexpectedError'; + import { ITelemetryService, noopTelemetryClient, @@ -6,15 +20,16 @@ import { type TelemetryProperties, type TelemetryServiceOptions, } from './telemetry'; -import { InstantiationType } from '#/_base/di/extensions'; -import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; export class TelemetryService implements ITelemetryService { - private client: TelemetryClient; + declare readonly _serviceBrand: undefined; + + private sinks: TelemetryClient[]; private context: TelemetryProperties; + private enabled = true; constructor(options: TelemetryServiceOptions = {}) { - this.client = options.client ?? noopTelemetryClient; + this.sinks = resolveSinks(options); this.context = { ...options.context, ...definedContext({ @@ -25,24 +40,67 @@ export class TelemetryService implements ITelemetryService { }; } - setDelegate(client: TelemetryClient): void { - this.client = client; - } - track(event: string, properties?: TelemetryProperties): void { - this.client.track(event, { ...this.context, ...properties }); + if (!this.enabled) { + return; + } + const merged = { ...this.context, ...properties }; + for (const sink of this.sinks) { + try { + sink.track(event, merged); + } catch (err) { + onUnexpectedError(err); + } + } } withContext(patch: TelemetryContextPatch): ITelemetryService { - return new TelemetryService({ - client: this.client.withContext?.(patch) ?? this.client, + const child = new TelemetryService({ + clients: this.sinks.map((sink) => sink.withContext?.(patch) ?? sink), context: { ...this.context, ...patch }, }); + child.enabled = this.enabled; + return child; } setContext(patch: TelemetryContextPatch): void { this.context = { ...this.context, ...patch }; - this.client.setContext?.(patch); + for (const sink of this.sinks) { + sink.setContext?.(patch); + } + } + + addSink(client: TelemetryClient): IDisposable { + this.sinks.push(client); + return toDisposable(() => this.removeSink(client)); + } + + removeSink(client: TelemetryClient): void { + this.sinks = this.sinks.filter((sink) => sink !== client); + } + + setDelegate(client: TelemetryClient): void { + this.sinks = [client]; + } + + setEnabled(enabled: boolean): void { + this.enabled = enabled; + } + + async flush(): Promise { + await Promise.all( + this.sinks.map((sink) => + Promise.resolve(sink.flush?.()).catch(onUnexpectedError), + ), + ); + } + + async shutdown(): Promise { + await Promise.all( + this.sinks.map((sink) => + Promise.resolve(sink.shutdown?.()).catch(onUnexpectedError), + ), + ); } } @@ -54,6 +112,13 @@ registerScopedService( 'telemetry', ); +function resolveSinks(options: TelemetryServiceOptions): TelemetryClient[] { + if (options.clients !== undefined) { + return options.clients.length > 0 ? [...options.clients] : [noopTelemetryClient]; + } + return [options.client ?? noopTelemetryClient]; +} + function definedContext(input: TelemetryProperties): TelemetryProperties { const out: Record> = {}; for (const [key, value] of Object.entries(input)) { diff --git a/packages/agent-core-v2/src/tool/toolService.ts b/packages/agent-core-v2/src/tool/toolService.ts index 5726ae4ae..f2ee8a706 100644 --- a/packages/agent-core-v2/src/tool/toolService.ts +++ b/packages/agent-core-v2/src/tool/toolService.ts @@ -16,11 +16,11 @@ import { type ServicesAccessor, } from '#/_base/di/instantiation'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { IAgentConfigService } from '#/config'; -import { IKaosService } from '#/kaos'; -import { ILLMService } from '#/kosong'; -import { IPermissionService } from '#/permission'; -import { IAgentRecords } from '#/records'; +import { ISessionConfigService } from '#/config/config'; +import { IKaosService } from '#/kaos/kaos'; +import { ILLMService } from '#/kosong/kosong'; +import { IPermissionService } from '#/permission/permission'; +import { IAgentRecords } from '#/records/records'; import { type ToolCallResult, @@ -67,7 +67,7 @@ export class ToolService implements IToolService { constructor( @IToolDefinitionRegistry private readonly registry: IToolDefinitionRegistry, - @IAgentConfigService _agentConfig: IAgentConfigService, + @ISessionConfigService _sessionConfig: ISessionConfigService, @IAgentRecords _records: IAgentRecords, @IKaosService _agentKaos: IKaosService, @IPermissionService _permission: IPermissionService, diff --git a/packages/agent-core-v2/src/wireRecord/wireRecord.ts b/packages/agent-core-v2/src/wireRecord/wireRecord.ts index da2c69413..8ce058d19 100644 --- a/packages/agent-core-v2/src/wireRecord/wireRecord.ts +++ b/packages/agent-core-v2/src/wireRecord/wireRecord.ts @@ -1,7 +1,7 @@ import type { ContentPart } from '@moonshot-ai/kosong'; -import { createDecorator } from "#/_base/di"; -import type { IDisposable } from "#/_base/di"; +import { createDecorator } from "#/_base/di/instantiation"; +import type { IDisposable } from "#/_base/di/lifecycle"; import type { IBlobStoreService } from '#/blobStore'; import type { Hooks } from '#/hooks'; diff --git a/packages/agent-core-v2/src/workspace/index.ts b/packages/agent-core-v2/src/workspace/index.ts index 5e20ff364..2d24c1d89 100644 --- a/packages/agent-core-v2/src/workspace/index.ts +++ b/packages/agent-core-v2/src/workspace/index.ts @@ -1,8 +1,7 @@ /** * `workspace` domain barrel — re-exports the workspace contract (`workspace`) - * and its scoped services (`workspaceService`). Importing this barrel registers - * the `IWorkspaceRegistry` and `IWorkspaceFsService` bindings into the scope - * registry. + * and its scoped service (`workspaceService`). Importing this barrel registers + * the `IWorkspaceService` binding into the scope registry. */ export * from './workspace'; diff --git a/packages/agent-core-v2/src/workspace/workspace.ts b/packages/agent-core-v2/src/workspace/workspace.ts index c467e6e44..5e0febcf8 100644 --- a/packages/agent-core-v2/src/workspace/workspace.ts +++ b/packages/agent-core-v2/src/workspace/workspace.ts @@ -1,33 +1,53 @@ /** - * `workspace` domain (cross-cutting) — core-scope workspace registry + fs. + * `workspace` domain (cross-cutting) — session-scope workspace service. * - * Defines the public contracts of workspace management: the `WorkspaceInfo` - * model, the `IWorkspaceRegistry` used to register and look up workspaces, and - * the `IWorkspaceFsService` used to resolve paths within a workspace. - * Core-scoped — shared across the application. + * Defines the public contract of the current session's workspace: the primary + * `workDir` plus any `additionalDirs`, and the path helpers built on top of + * them (`resolve` / `isWithin` / `assertAllowed`). Session-scoped — one + * instance per session — so consumers never thread a `workspaceId` around. + * + * This is the semantic layer above `kaos`: `ISessionKaosService` owns the raw + * `Kaos` environments and additional roots, while `IWorkspaceService` turns + * them into workspace-relative path operations. Dependency direction is + * `workspace → kaos`, never the reverse (see `docs/service-design.md` §5). */ import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { PathAccessOperation } from '#/_base/tools/policies/path-access'; +import type { WorkspaceConfig } from '#/_base/tools/support/workspace'; -export interface WorkspaceInfo { - readonly id: string; - readonly root: string; -} - -export interface IWorkspaceRegistry { +export interface IWorkspaceService { readonly _serviceBrand: undefined; - register(root: string): WorkspaceInfo; - get(id: string): WorkspaceInfo | undefined; - list(): readonly WorkspaceInfo[]; + + /** Primary workspace directory (absolute). Reflects the session kaos cwd. */ + readonly workDir: string; + /** Extra allowed roots (e.g. `--add-dir` CLI flag). */ + readonly additionalDirs: readonly string[]; + + /** + * Resolve `rel` to an absolute, canonical path. Relative paths are resolved + * against `workDir`; absolute paths are normalized as-is. Lexical only — no + * filesystem I/O. + */ + resolve(rel: string): string; + + /** True iff `path` (resolved against `workDir`) sits inside `workDir` or any `additionalDirs`. */ + isWithin(path: string): boolean; + + /** + * Throw if `path` escapes the workspace for `op` (or matches a sensitive + * file); returns the canonical absolute path when it passes. + */ + assertAllowed(path: string, op: PathAccessOperation): string; + + /** Snapshot as a `WorkspaceConfig` value object for the path-access helpers. */ + toConfig(): WorkspaceConfig; + + /** Add an extra allowed root for this session (no-op if already present). */ + addAdditionalDir(dir: string): void; + /** Remove an extra allowed root from this session. */ + removeAdditionalDir(dir: string): void; } -export const IWorkspaceRegistry: ServiceIdentifier = - createDecorator('workspaceRegistry'); - -export interface IWorkspaceFsService { - readonly _serviceBrand: undefined; - resolve(workspaceId: string, rel: string): string; -} - -export const IWorkspaceFsService: ServiceIdentifier = - createDecorator('workspaceFsService'); +export const IWorkspaceService: ServiceIdentifier = + createDecorator('workspaceService'); diff --git a/packages/agent-core-v2/src/workspace/workspaceService.ts b/packages/agent-core-v2/src/workspace/workspaceService.ts index d014842d5..4873e2571 100644 --- a/packages/agent-core-v2/src/workspace/workspaceService.ts +++ b/packages/agent-core-v2/src/workspace/workspaceService.ts @@ -1,64 +1,87 @@ /** - * `workspace` domain (cross-cutting) — `IWorkspaceRegistry` / - * `IWorkspaceFsService` implementation. + * `workspace` domain (cross-cutting) — `IWorkspaceService` implementation. * - * Owns the workspace registry and path resolution; resolves filesystem access - * through `kaos` and logs through `log`. Bound at Core scope. + * Turns the raw `Kaos` environments and additional roots owned by + * `ISessionKaosService` into workspace-relative path operations. Bound at + * Session scope — one instance per session. */ -import { join } from 'node:path'; +import type { Kaos } from '@moonshot-ai/kaos'; import { InstantiationType } from '#/_base/di/extensions'; +import { Disposable } from '#/_base/di/lifecycle'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { IKaosFactory } from '#/kaos'; -import { ILogService } from '#/log'; - import { - type WorkspaceInfo, - IWorkspaceFsService, - IWorkspaceRegistry, -} from './workspace'; + type PathAccessOperation, + type PathClass, + assertPathAllowed, + canonicalizePath, + isWithinWorkspace, +} from '#/_base/tools/policies/path-access'; +import type { WorkspaceConfig } from '#/_base/tools/support/workspace'; +import { ISessionKaosService } from '#/kaos/kaos'; +import { ILogService } from '#/log/log'; -let nextWorkspaceId = 0; +import { IWorkspaceService } from './workspace'; -export class WorkspaceRegistry implements IWorkspaceRegistry { - declare readonly _serviceBrand: undefined; - private readonly workspaces = new Map(); - - constructor( - @IKaosFactory _kaosFactory: IKaosFactory, - @ILogService _log: ILogService, - ) {} - - register(root: string): WorkspaceInfo { - const id = `ws-${nextWorkspaceId++}`; - const info: WorkspaceInfo = { id, root }; - this.workspaces.set(id, info); - return info; - } - get(id: string): WorkspaceInfo | undefined { - return this.workspaces.get(id); - } - list(): readonly WorkspaceInfo[] { - return [...this.workspaces.values()]; - } -} - -export class WorkspaceFsService implements IWorkspaceFsService { +export class WorkspaceService extends Disposable implements IWorkspaceService { declare readonly _serviceBrand: undefined; constructor( - private readonly registry: IWorkspaceRegistry = new WorkspaceRegistry(undefined as never, undefined as never), - @IKaosFactory _kaosFactory: IKaosFactory, - @ILogService _log: ILogService, - ) {} + @ISessionKaosService private readonly sessionKaos: ISessionKaosService, + @ILogService private readonly _log: ILogService, + ) { + super(); + } - resolve(workspaceId: string, rel: string): string { - const ws = this.registry.get(workspaceId); - if (ws === undefined) throw new Error(`unknown workspace '${workspaceId}'`); - return join(ws.root, rel); + private get kaos(): Kaos { + return this.sessionKaos.toolKaos; + } + + private get pathClass(): PathClass { + return this.kaos.pathClass(); + } + + get workDir(): string { + return this.kaos.getcwd(); + } + + get additionalDirs(): readonly string[] { + return this.sessionKaos.additionalDirs; + } + + resolve(rel: string): string { + return canonicalizePath(rel, this.workDir, this.pathClass); + } + + isWithin(path: string): boolean { + return isWithinWorkspace(this.resolve(path), this.toConfig(), this.pathClass); + } + + assertAllowed(path: string, op: PathAccessOperation): string { + return assertPathAllowed(path, this.workDir, this.toConfig(), { + mode: op, + pathClass: this.pathClass, + }); + } + + toConfig(): WorkspaceConfig { + return { workspaceDir: this.workDir, additionalDirs: this.additionalDirs }; + } + + addAdditionalDir(dir: string): void { + this.sessionKaos.addAdditionalDir(dir); + } + + removeAdditionalDir(dir: string): void { + this.sessionKaos.removeAdditionalDir(dir); } } -registerScopedService(LifecycleScope.Core, IWorkspaceRegistry, WorkspaceRegistry, InstantiationType.Delayed, 'workspace'); -registerScopedService(LifecycleScope.Core, IWorkspaceFsService, WorkspaceFsService, InstantiationType.Delayed, 'workspace'); +registerScopedService( + LifecycleScope.Session, + IWorkspaceService, + WorkspaceService, + InstantiationType.Delayed, + 'workspace', +); diff --git a/packages/agent-core-v2/test/auth/auth.test.ts b/packages/agent-core-v2/test/auth/auth.test.ts index 33e0c791e..c2756a189 100644 --- a/packages/agent-core-v2/test/auth/auth.test.ts +++ b/packages/agent-core-v2/test/auth/auth.test.ts @@ -1,40 +1,225 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +/** + * `auth` domain tests — covers the `OAuthService` device-code orchestration + * and its dependency on the `providers` config section, using a fake + * `KimiOAuthToolkit` so no real network or token storage is exercised. + */ + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import type { KimiOAuthToolkit } from '@moonshot-ai/kimi-code-oauth'; import { DisposableStore } from '#/_base/di/lifecycle'; -import { createServices } from '#/_base/di/test'; -import type { TestInstantiationService } from '#/_base/di/test'; -import { IOAuthService } from '#/auth/auth'; +import { createServices, type TestInstantiationService } from '#/_base/di/test'; +import { ErrorCodes } from '#/_base/errors/codes'; +import { KimiError } from '#/_base/errors/errors'; +import { IAuthSummaryService, IOAuthService } from '#/auth/auth'; +import { AuthSummaryService, OAuthService } from '#/auth/authService'; +import { PROVIDERS_SECTION, type ProvidersSection } from '#/auth/oauthSchemas'; +import { IConfigService } from '#/config/config'; -import { OAuthService } from '#/auth/authService'; import { registerConfigServices } from '../config/stubs'; import { registerEnvironmentServices } from '../environment/stubs'; import { registerTelemetryServices } from '../telemetry/stubs'; +const OAUTH_PROVIDER = 'managed:kimi-code'; +const NON_OAUTH_PROVIDER = 'openai-main'; + +const deviceAuth = { + userCode: 'ABCD-EFGH', + deviceCode: 'device-code', + verificationUri: 'https://example.com/device', + verificationUriComplete: 'https://example.com/device?code=ABCD-EFGH', + expiresIn: 900, + interval: 5, +}; + +const flush = (): Promise => new Promise((resolve) => setTimeout(resolve, 0)); + +interface FakeToolkit { + readonly login: ReturnType; + readonly logout: ReturnType; + readonly getCachedAccessToken: ReturnType; + readonly tokenProvider: ReturnType; +} + describe('OAuthService', () => { let disposables: DisposableStore; let ix: TestInstantiationService; + let providers: ProvidersSection; + let toolkit: FakeToolkit; beforeEach(() => { disposables = new DisposableStore(); + providers = { + [OAUTH_PROVIDER]: { + type: 'kimi', + baseUrl: 'https://api.example.com', + oauth: { storage: 'file', key: 'oauth/kimi-code' }, + }, + [NON_OAUTH_PROVIDER]: { type: 'openai', apiKey: 'sk-test' }, + }; ix = createServices(disposables, { - base: [ - registerConfigServices, - registerEnvironmentServices, - registerTelemetryServices, - ], + base: [registerConfigServices, registerEnvironmentServices, registerTelemetryServices], additionalServices: (reg) => { - reg.define(IOAuthService, OAuthService); + reg.definePartialInstance(IConfigService, { + get: ((domain: string) => + domain === PROVIDERS_SECTION ? providers : undefined) as IConfigService['get'], + onDidChange: (() => ({ dispose: () => {} })) as IConfigService['onDidChange'], + }); + }, + }); + toolkit = { + login: vi.fn(), + logout: vi.fn().mockResolvedValue({ providerName: OAUTH_PROVIDER, ok: true }), + getCachedAccessToken: vi.fn().mockResolvedValue(undefined), + tokenProvider: vi.fn().mockReturnValue({ getAccessToken: async () => 'access-token' }), + }; + }); + afterEach(() => disposables.dispose()); + + function createService(): IOAuthService { + return ix.createInstance(OAuthService, toolkit as unknown as KimiOAuthToolkit); + } + + it('startLogin resolves a device-code flow and flips to authenticated on success', async () => { + toolkit.login.mockImplementation(async (_provider, options) => { + options.onDeviceCode(deviceAuth); + return { providerName: OAUTH_PROVIDER, ok: true }; + }); + const svc = createService(); + + const start = await svc.startLogin(OAUTH_PROVIDER); + expect(start).toMatchObject({ + provider: OAUTH_PROVIDER, + verification_uri: deviceAuth.verificationUri, + verification_uri_complete: deviceAuth.verificationUriComplete, + user_code: deviceAuth.userCode, + interval: deviceAuth.interval, + status: 'pending', + }); + expect(toolkit.login).toHaveBeenCalledWith( + OAUTH_PROVIDER, + expect.objectContaining({ oauthRef: { storage: 'file', key: 'oauth/kimi-code' } }), + ); + + await flush(); + expect(svc.getFlow(OAUTH_PROVIDER)?.status).toBe('authenticated'); + }); + + it('startLogin rejects with AUTH_LOGIN_REQUIRED when provider has no oauth config', async () => { + const svc = createService(); + await expect(svc.startLogin(NON_OAUTH_PROVIDER)).rejects.toThrow(KimiError); + await expect(svc.startLogin(NON_OAUTH_PROVIDER)).rejects.toMatchObject({ + code: ErrorCodes.AUTH_LOGIN_REQUIRED, + }); + expect(toolkit.login).not.toHaveBeenCalled(); + }); + + it('cancelLogin aborts a pending flow and marks it cancelled', async () => { + let capturedSignal: AbortSignal | undefined; + toolkit.login.mockImplementation(async (_provider, options) => { + capturedSignal = options.signal; + options.onDeviceCode(deviceAuth); + return new Promise(() => {}); + }); + const svc = createService(); + await svc.startLogin(OAUTH_PROVIDER); + + const result = await svc.cancelLogin(OAUTH_PROVIDER); + expect(result).toEqual({ cancelled: true, status: 'cancelled' }); + expect(capturedSignal?.aborted).toBe(true); + expect(svc.getFlow(OAUTH_PROVIDER)?.status).toBe('cancelled'); + }); + + it('logout delegates to the toolkit and clears any pending flow', async () => { + toolkit.login.mockImplementation(async (_provider, options) => { + options.onDeviceCode(deviceAuth); + return new Promise(() => {}); + }); + const svc = createService(); + await svc.startLogin(OAUTH_PROVIDER); + + const result = await svc.logout(OAUTH_PROVIDER); + expect(result).toEqual({ logged_out: true, provider: OAUTH_PROVIDER }); + expect(toolkit.logout).toHaveBeenCalledWith( + OAUTH_PROVIDER, + { storage: 'file', key: 'oauth/kimi-code' }, + ); + }); + + it('status reports loggedIn based on the cached access token', async () => { + const svc = createService(); + expect(await svc.status(OAUTH_PROVIDER)).toEqual({ loggedIn: false }); + + toolkit.getCachedAccessToken.mockResolvedValue('cached-token'); + expect(await svc.status(OAUTH_PROVIDER)).toEqual({ + loggedIn: true, + provider: OAUTH_PROVIDER, + }); + }); + + it('resolveTokenProvider delegates to the toolkit', () => { + const svc = createService(); + const provider = svc.resolveTokenProvider(OAUTH_PROVIDER, { storage: 'file', key: 'k' }); + expect(provider).toEqual({ getAccessToken: expect.any(Function) }); + expect(toolkit.tokenProvider).toHaveBeenCalledWith(OAUTH_PROVIDER, { + storage: 'file', + key: 'k', + }); + }); +}); + +describe('AuthSummaryService', () => { + let disposables: DisposableStore; + let ix: TestInstantiationService; + let providers: ProvidersSection; + let oauthStatus: ReturnType; + + beforeEach(() => { + disposables = new DisposableStore(); + providers = { + [OAUTH_PROVIDER]: { + type: 'kimi', + oauth: { storage: 'file', key: 'oauth/kimi-code' }, + }, + [NON_OAUTH_PROVIDER]: { type: 'openai', apiKey: 'sk-test' }, + }; + oauthStatus = vi.fn(); + ix = createServices(disposables, { + additionalServices: (reg) => { + reg.definePartialInstance(IConfigService, { + get: ((domain: string) => + domain === PROVIDERS_SECTION ? providers : undefined) as IConfigService['get'], + }); + reg.definePartialInstance(IOAuthService, { + status: oauthStatus as unknown as IOAuthService['status'], + }); }, }); }); afterEach(() => disposables.dispose()); - it('login / status / logout', async () => { - const svc = ix.get(IOAuthService); - expect(await svc.status()).toEqual({ loggedIn: false }); - await svc.login('kimi'); - expect(await svc.status()).toEqual({ loggedIn: true, provider: 'kimi' }); - await svc.logout('kimi'); - expect(await svc.status()).toEqual({ loggedIn: false }); + function createSummary(): IAuthSummaryService { + return ix.createInstance(AuthSummaryService); + } + + it('summarize reports status only for providers configured with oauth', async () => { + oauthStatus.mockResolvedValue({ loggedIn: true, provider: OAUTH_PROVIDER }); + const result = await createSummary().summarize(); + expect(result).toEqual([{ loggedIn: true, provider: OAUTH_PROVIDER }]); + expect(oauthStatus).toHaveBeenCalledWith(OAUTH_PROVIDER); + expect(oauthStatus).not.toHaveBeenCalledWith(NON_OAUTH_PROVIDER); + }); + + it('ensureReady rejects with AUTH_LOGIN_REQUIRED when the provider is logged out', async () => { + oauthStatus.mockResolvedValue({ loggedIn: false }); + await expect(createSummary().ensureReady(OAUTH_PROVIDER)).rejects.toMatchObject({ + code: ErrorCodes.AUTH_LOGIN_REQUIRED, + }); + }); + + it('ensureReady resolves when the provider is logged in', async () => { + oauthStatus.mockResolvedValue({ loggedIn: true, provider: OAUTH_PROVIDER }); + await expect(createSummary().ensureReady(OAUTH_PROVIDER)).resolves.toBeUndefined(); }); }); diff --git a/packages/agent-core-v2/test/config/config.test.ts b/packages/agent-core-v2/test/config/config.test.ts index 293bf2478..b1e4e0085 100644 --- a/packages/agent-core-v2/test/config/config.test.ts +++ b/packages/agent-core-v2/test/config/config.test.ts @@ -1,121 +1,208 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import { IKaosService } from '#/kaos/kaos'; +import { mkdtemp, readFile, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { z } from 'zod'; import { DisposableStore } from '#/_base/di/lifecycle'; import { createServices } from '#/_base/di/test'; import type { TestInstantiationService } from '#/_base/di/test'; -import { IAgentConfigService, IConfigService } from '#/config/config'; - -import { AgentConfigService, ConfigRegistry, ConfigService } from '#/config/configService'; +import { IEnvironmentService } from '#/environment/environment'; +import { + IConfigRegistry, + IConfigService, + ISessionConfigService, + type SessionConfigSection, +} from '#/config/config'; +import { ConfigRegistry, ConfigService } from '#/config/configService'; +import { SessionConfigService } from '#/config/sessionConfigService'; +import { ISessionMetaStore } from '#/records/records'; import { registerConfigServices } from '../config/stubs'; -import { registerEnvironmentServices } from '../environment/stubs'; +import { stubEnvironment } from '../environment/stubs'; import { registerLogServices } from '../log/stubs'; -import { registerRecordsServices } from '../records/stubs'; + +const passthroughSchema = { parse: (value: unknown) => value }; + +function stubSessionMetaStore(initial: Record = {}) { + let data = { ...initial }; + const store: ISessionMetaStore = { + _serviceBrand: undefined, + read: () => Promise.resolve({ ...data }), + write: (patch) => { + data = { ...data, ...patch }; + return Promise.resolve(); + }, + flush: () => Promise.resolve(), + }; + return { + store, + snapshot() { + return { ...data }; + }, + }; +} describe('ConfigRegistry', () => { it('registers and retrieves a section', () => { const reg = new ConfigRegistry(); - const schema = { type: 'object' }; - reg.registerSection('permission', schema); - expect(reg.getSection('permission')).toEqual({ domain: 'permission', schema }); + reg.registerSection('permission', passthroughSchema); + expect(reg.getSection('permission')).toMatchObject({ domain: 'permission' }); expect(reg.getSection('missing')).toBeUndefined(); }); it('throws when the same domain is registered twice', () => { const reg = new ConfigRegistry(); - reg.registerSection('permission', { type: 'object' }); - expect(() => reg.registerSection('permission', { type: 'object' })).toThrow( - /already registered/, - ); + reg.registerSection('permission', passthroughSchema); + expect(() => reg.registerSection('permission', passthroughSchema)).toThrow(/already registered/); }); it('deep-merges patches', () => { const reg = new ConfigRegistry(); - const merged = reg.merge({ a: 1, nested: { x: 1, y: 2 } }, { nested: { y: 3, z: 4 }, b: 2 }); + const merged = reg.merge('session', { a: 1, nested: { x: 1, y: 2 } }, { nested: { y: 3, z: 4 }, b: 2 }); expect(merged).toEqual({ a: 1, b: 2, nested: { x: 1, y: 3, z: 4 } }); }); + + it('validates with the registered schema', () => { + const reg = new ConfigRegistry(); + reg.registerSection('session', z.object({ modelAlias: z.string() })); + expect(reg.validate('session', { modelAlias: 'k2' })).toEqual({ modelAlias: 'k2' }); + expect(() => reg.validate('session', { modelAlias: 1 })).toThrow(); + }); + + it('returns registered defaults', () => { + const reg = new ConfigRegistry(); + reg.registerSection('session', passthroughSchema, { defaultValue: { modelAlias: 'default' } }); + expect(reg.defaultValue('session')).toEqual({ modelAlias: 'default' }); + }); }); describe('ConfigService', () => { let disposables: DisposableStore; let ix: TestInstantiationService; + let homeDir: string; - beforeEach(() => { + beforeEach(async () => { disposables = new DisposableStore(); + homeDir = await mkdtemp(join(tmpdir(), 'kimi-config-')); ix = createServices(disposables, { - base: [ - registerConfigServices, - registerEnvironmentServices, - registerLogServices, - ], + base: [registerConfigServices, registerLogServices], additionalServices: (reg) => { + reg.defineInstance(IEnvironmentService, stubEnvironment(homeDir)); reg.define(IConfigService, ConfigService); }, }); }); - afterEach(() => disposables.dispose()); + afterEach(async () => { + disposables.dispose(); + await rm(homeDir, { recursive: true, force: true }); + }); it('set merges and get reads back', async () => { const svc = ix.get(IConfigService); - await svc.set('agent', { modelAlias: 'k2', nested: { a: 1 } }); - await svc.set('agent', { nested: { b: 2 } }); - expect(svc.get('agent')).toEqual({ modelAlias: 'k2', nested: { a: 1, b: 2 } }); + await svc.set('session', { modelAlias: 'k2', nested: { a: 1 } }); + await svc.set('session', { nested: { b: 2 } }); + expect(svc.get('session')).toEqual({ modelAlias: 'k2', nested: { a: 1, b: 2 } }); + }); + + it('persists config to config.toml', async () => { + const svc = ix.get(IConfigService); + await svc.set('session', { modelAlias: 'k2' }); + const text = await readFile(join(homeDir, 'config.toml'), 'utf-8'); + expect(text).toContain('[session]'); + expect(text).toContain('modelAlias = "k2"'); + }); + + it('reloads config from disk', async () => { + const svc = ix.get(IConfigService); + await svc.set('session', { modelAlias: 'k2' }); + await svc.reload(); + expect(svc.get('session')).toEqual({ modelAlias: 'k2' }); }); it('fires onDidChange with the domain', async () => { const svc = ix.get(IConfigService); const fired: string[] = []; disposables.add(svc.onDidChange((e) => fired.push(e.domain))); - await svc.set('agent', { modelAlias: 'k2' }); + await svc.set('session', { modelAlias: 'k2' }); await svc.set('tool', { x: 1 }); - expect(fired).toEqual(['agent', 'tool']); + expect(fired).toEqual(['session', 'tool']); + }); + + it('rejects invalid patches and does not write them', async () => { + const registry = ix.get(IConfigRegistry); + registry.registerSection('session', z.object({ modelAlias: z.string() })); + const svc = ix.get(IConfigService); + await expect(svc.set('session', { modelAlias: 1 })).rejects.toThrow(); + await expect(readFile(join(homeDir, 'config.toml'), 'utf-8')).rejects.toThrow(); }); }); -describe('AgentConfigService', () => { +describe('SessionConfigService', () => { let disposables: DisposableStore; let ix: TestInstantiationService; - let agentSection: Record; - - const agentKaos: IKaosService = { - _serviceBrand: undefined, - get kaos(): never { - throw new Error('unused'); - }, - cwd: '/repo', - chdir: () => Promise.resolve(), - }; + let sessionSection: SessionConfigSection; + let metaStore: ReturnType; beforeEach(() => { disposables = new DisposableStore(); - agentSection = {}; + sessionSection = {}; + metaStore = stubSessionMetaStore(); ix = createServices(disposables, { - base: [registerRecordsServices], + base: [registerLogServices], additionalServices: (reg) => { - reg.definePartialInstance(IConfigService, { get: () => agentSection as T }); - reg.defineInstance(IKaosService, agentKaos); - reg.define(IAgentConfigService, AgentConfigService); + reg.definePartialInstance(IConfigService, { get: () => sessionSection as T }); + reg.defineInstance(ISessionMetaStore, metaStore.store); + reg.define(ISessionConfigService, SessionConfigService); }, }); }); afterEach(() => disposables.dispose()); - it('reads the agent section and cwd from kaos', () => { - agentSection = { modelAlias: 'k2', systemPrompt: 'hi', provider: 'p' }; - const view = ix.get(IAgentConfigService); + it('reads the session section from global config', async () => { + sessionSection = { modelAlias: 'k2', systemPrompt: 'hi', provider: 'p' }; + const view = ix.get(ISessionConfigService); + await view.ready; expect(view.modelAlias).toBe('k2'); expect(view.systemPrompt).toBe('hi'); expect(view.provider).toBe('p'); expect(view.thinkingLevel).toBeUndefined(); - expect(view.cwd).toBe('/repo'); }); - it('setModel / setThinking update the view', async () => { - const view = ix.get(IAgentConfigService); + it('restores session metadata overrides', async () => { + metaStore = stubSessionMetaStore({ modelAlias: 'restored', thinkingLevel: 'high' }); + ix = createServices(disposables, { + base: [registerLogServices], + additionalServices: (reg) => { + reg.definePartialInstance(IConfigService, { get: () => sessionSection as T }); + reg.defineInstance(ISessionMetaStore, metaStore.store); + reg.define(ISessionConfigService, SessionConfigService); + }, + }); + const view = ix.get(ISessionConfigService); + await view.ready; + expect(view.modelAlias).toBe('restored'); + expect(view.thinkingLevel).toBe('high'); + }); + + it('setModel / setThinking persist metadata and update the view', async () => { + const view = ix.get(ISessionConfigService); + await view.ready; await view.setModel('k1'); await view.setThinking('high'); expect(view.modelAlias).toBe('k1'); expect(view.thinkingLevel).toBe('high'); + expect(metaStore.snapshot()).toEqual({ modelAlias: 'k1', thinkingLevel: 'high' }); + }); + + it('fires onDidChange after updates', async () => { + const view = ix.get(ISessionConfigService); + await view.ready; + const changed: string[] = []; + disposables.add(view.onDidChange((e) => changed.push(...e.changed))); + await view.setModel('k1'); + await view.setThinking('high'); + expect(changed).toEqual(['modelAlias', 'thinkingLevel']); }); }); diff --git a/packages/agent-core-v2/test/config/stubs.ts b/packages/agent-core-v2/test/config/stubs.ts index 3df3a6e27..ff7dc56f0 100644 --- a/packages/agent-core-v2/test/config/stubs.ts +++ b/packages/agent-core-v2/test/config/stubs.ts @@ -6,17 +6,17 @@ */ import type { ServiceRegistration } from '#/_base/di/test'; -import { IAgentConfigService, IConfigRegistry, IConfigService } from '#/config/config'; +import { IConfigRegistry, IConfigService, ISessionConfigService } from '#/config/config'; import { ConfigRegistry } from '#/config/configService'; /** * Register the default config collaborators: a real `ConfigRegistry` plus empty - * `IConfigService` / `IAgentConfigService` placeholders. Tests exercising the - * real `ConfigService` / `AgentConfigService` should override the placeholder + * `IConfigService` / `ISessionConfigService` placeholders. Tests exercising the + * real `ConfigService` / `SessionConfigService` should override the placeholder * via `additionalServices`. */ export function registerConfigServices(reg: ServiceRegistration): void { reg.defineInstance(IConfigRegistry, new ConfigRegistry()); reg.definePartialInstance(IConfigService, {}); - reg.definePartialInstance(IAgentConfigService, {}); + reg.definePartialInstance(ISessionConfigService, {}); } diff --git a/packages/agent-core-v2/test/environment/stubs.ts b/packages/agent-core-v2/test/environment/stubs.ts index de8f40437..dbb77a900 100644 --- a/packages/agent-core-v2/test/environment/stubs.ts +++ b/packages/agent-core-v2/test/environment/stubs.ts @@ -23,7 +23,8 @@ export function stubEnvironment(homeDir = '/tmp/kimi-home'): IEnvironmentService }; } -/** Register the default `IEnvironmentService` rooted at `/tmp/kimi-home`. */ +/** Register the default `IEnvironmentService` rooted at an isolated temp dir. */ export function registerEnvironmentServices(reg: ServiceRegistration): void { - reg.defineInstance(IEnvironmentService, stubEnvironment()); + const homeDir = `/tmp/kimi-code-agent-core-v2-${process.pid}-${Date.now()}-${Math.random().toString(36).slice(2)}`; + reg.defineInstance(IEnvironmentService, stubEnvironment(homeDir)); } diff --git a/packages/agent-core-v2/test/gateway/gateway.test.ts b/packages/agent-core-v2/test/gateway/gateway.test.ts index 1f3270d5c..d2d20243a 100644 --- a/packages/agent-core-v2/test/gateway/gateway.test.ts +++ b/packages/agent-core-v2/test/gateway/gateway.test.ts @@ -6,8 +6,10 @@ import { type IScopeHandle, LifecycleScope } from '#/_base/di/scope'; import { createServices } from '#/_base/di/test'; import type { TestInstantiationService } from '#/_base/di/test'; import { IAgentLifecycleService } from '#/agent-lifecycle/agentLifecycle'; -import { IRestGateway, IScopeRegistry } from '#/gateway'; +import { IRestGateway, IScopeRegistry } from '#/gateway/gateway'; import { RestGateway, ScopeRegistry } from '#/gateway/gatewayService'; +import { ILogService, ISessionLogService } from '#/log/log'; +import { stubLog } from '../log/stubs'; import { stubTurn } from '../turn/stubs'; describe('ScopeRegistry', () => { @@ -40,6 +42,7 @@ describe('RestGateway', () => { const ix = createServices(disposables, { additionalServices: (reg) => { reg.define(IRestGateway, RestGateway); + reg.defineInstance(ILogService, stubLog()); }, }); @@ -76,3 +79,65 @@ describe('RestGateway', () => { disposables.dispose(); }); }); + +describe('RestGateway.flushLogs', () => { + function buildGateway(opts: { + sessionLog?: { flush: () => Promise }; + globalLog?: ILogService; + sessionId?: string; + }) { + const disposables = new DisposableStore(); + const ix = createServices(disposables, { + additionalServices: (reg) => { + reg.define(IRestGateway, RestGateway); + reg.defineInstance(ILogService, opts.globalLog ?? stubLog()); + }, + }); + const sessionHandle: IScopeHandle = { + id: opts.sessionId ?? 's1', + kind: LifecycleScope.Session, + accessor: { + get: (id: unknown) => (id === ISessionLogService ? opts.sessionLog : undefined), + } as unknown as ServicesAccessor, + }; + ix.stub(IScopeRegistry, { + _serviceBrand: undefined, + createSession: () => Promise.resolve(sessionHandle), + get: (id) => (id === (opts.sessionId ?? 's1') ? sessionHandle : undefined), + close: () => Promise.resolve(), + }); + return { gw: ix.get(IRestGateway), disposables }; + } + + it('flushes the session log service for a known session', async () => { + let flushed = false; + const { gw, disposables } = buildGateway({ + sessionLog: { flush: () => { flushed = true; return Promise.resolve(); } }, + }); + await gw.flushLogs('s1'); + expect(flushed).toBe(true); + disposables.dispose(); + }); + + it('is a no-op for an unknown session', async () => { + let flushed = false; + const { gw, disposables } = buildGateway({ + sessionLog: { flush: () => { flushed = true; return Promise.resolve(); } }, + }); + await expect(gw.flushLogs('nope')).resolves.toBeUndefined(); + expect(flushed).toBe(false); + disposables.dispose(); + }); + + it('flushGlobalLogs delegates to the Core ILogService', async () => { + let flushed = false; + const globalLog: ILogService = { + ...stubLog(), + flush: () => { flushed = true; return Promise.resolve(); }, + }; + const { gw, disposables } = buildGateway({ globalLog }); + await gw.flushGlobalLogs(); + expect(flushed).toBe(true); + disposables.dispose(); + }); +}); diff --git a/packages/agent-core-v2/test/lint/domain-layers.test.ts b/packages/agent-core-v2/test/lint/domain-layers.test.ts index db3adbbf4..14b1de542 100644 --- a/packages/agent-core-v2/test/lint/domain-layers.test.ts +++ b/packages/agent-core-v2/test/lint/domain-layers.test.ts @@ -35,7 +35,7 @@ describe('check-domain-layers', () => { it('flags a lower layer importing a higher layer', () => { const violations = checkSource( - `import { ITurnService } from '#/turn';`, + `import { ITurnService } from '#/turn/turn';`, at('log', 'log.ts'), ); expect(violations).toHaveLength(1); diff --git a/packages/agent-core-v2/test/telemetry/cloudSink.test.ts b/packages/agent-core-v2/test/telemetry/cloudSink.test.ts new file mode 100644 index 000000000..dc89715af --- /dev/null +++ b/packages/agent-core-v2/test/telemetry/cloudSink.test.ts @@ -0,0 +1,215 @@ +import { mkdtempSync, readdirSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import { CloudSink, type CloudSinkOptions } from '#/telemetry/cloudSink'; + +interface CapturedRequest { + readonly url: string; + readonly headers: Record; + readonly body: { + readonly user_id: string; + readonly events: readonly Record[]; + }; +} + +type Responder = (req: CapturedRequest) => Response | Promise; + +function makeFetch(responder: Responder): typeof fetch { + return (async (input: unknown, init: unknown) => { + const requestInit = init as { headers: Record; body: string }; + const req: CapturedRequest = { + url: String(input), + headers: requestInit.headers, + body: JSON.parse(requestInit.body) as CapturedRequest['body'], + }; + return responder(req); + }) as unknown as typeof fetch; +} + +function okResponse(): Response { + return new Response(null, { status: 200 }); +} + +function statusResponse(status: number): Response { + return new Response(null, { status }); +} + +function baseOptions(overrides: Partial = {}): CloudSinkOptions { + return { + homeDir: overrides.homeDir ?? '', + deviceId: overrides.deviceId ?? 'dev', + appName: overrides.appName ?? 'test-app', + version: overrides.version ?? '1.0.0', + env: overrides.env ?? {}, + sleep: overrides.sleep ?? (async () => {}), + ...overrides, + }; +} + +describe('CloudSink', () => { + let homeDir: string; + + beforeEach(() => { + homeDir = mkdtempSync(join(tmpdir(), 'cloud-sink-')); + }); + + afterEach(() => { + rmSync(homeDir, { recursive: true, force: true }); + }); + + it('sends a flattened, prefixed payload with user_id and context', async () => { + const requests: CapturedRequest[] = []; + const sink = new CloudSink( + baseOptions({ + homeDir, + deviceId: 'dev123', + sessionId: 'sess1', + fetchImpl: makeFetch((req) => { + requests.push(req); + return okResponse(); + }), + }), + ); + + sink.track('tool.call', { name: 'bash', count: 2 }); + await sink.flush(); + + expect(requests).toHaveLength(1); + expect(requests[0]?.url).toBe('https://telemetry-logs.kimi.com/v1/event'); + expect(requests[0]?.body.user_id).toBe('kfc_device_id_dev123'); + const event = requests[0]?.body.events[0]; + expect(event?.['event']).toBe('kfc_tool.call'); + expect(event?.['device_id']).toBe('dev123'); + expect(event?.['session_id']).toBe('sess1'); + expect(event?.['property_name']).toBe('bash'); + expect(event?.['property_count']).toBe(2); + expect(event?.['context_app_name']).toBe('test-app'); + expect(event?.['context_version']).toBe('1.0.0'); + expect(typeof event?.['event_id']).toBe('string'); + expect(typeof event?.['timestamp']).toBe('number'); + }); + + it('sends Authorization header when a token is provided', async () => { + const requests: CapturedRequest[] = []; + const sink = new CloudSink( + baseOptions({ + homeDir, + getAccessToken: () => 'tok123', + fetchImpl: makeFetch((req) => { + requests.push(req); + return okResponse(); + }), + }), + ); + + sink.track('evt'); + await sink.flush(); + + expect(requests[0]?.headers['Authorization']).toBe('Bearer tok123'); + }); + + it('auto-flushes when the buffer reaches the threshold', async () => { + let sends = 0; + const sink = new CloudSink( + baseOptions({ + homeDir, + flushThreshold: 3, + fetchImpl: makeFetch(() => { + sends += 1; + return okResponse(); + }), + }), + ); + + sink.track('e1'); + sink.track('e2'); + expect(sends).toBe(0); + sink.track('e3'); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(sends).toBe(1); + }); + + it('shutdown flushes the remaining buffered events', async () => { + let sends = 0; + const sink = new CloudSink( + baseOptions({ + homeDir, + fetchImpl: makeFetch(() => { + sends += 1; + return okResponse(); + }), + }), + ); + + sink.track('e1'); + await sink.shutdown(); + expect(sends).toBe(1); + }); + + it('retries on 5xx and saves to disk after exhausting backoffs', async () => { + let attempts = 0; + const sink = new CloudSink( + baseOptions({ + homeDir, + fetchImpl: makeFetch(() => { + attempts += 1; + return statusResponse(500); + }), + }), + ); + + sink.track('evt'); + await sink.flush(); + + expect(attempts).toBe(4); + const files = readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')); + expect(files).toHaveLength(1); + }); + + it('retries a 401 once without the Authorization header', async () => { + const seenAuths: (string | undefined)[] = []; + const sink = new CloudSink( + baseOptions({ + homeDir, + getAccessToken: () => 'tok', + fetchImpl: makeFetch((req) => { + seenAuths.push(req.headers['Authorization']); + if (req.headers['Authorization'] !== undefined) { + return statusResponse(401); + } + return okResponse(); + }), + }), + ); + + sink.track('evt'); + await sink.flush(); + + expect(seenAuths).toEqual(['Bearer tok', undefined]); + }); + + it('retryDiskEvents resends saved events and removes the file on success', async () => { + let shouldFail = true; + const sink = new CloudSink( + baseOptions({ + homeDir, + fetchImpl: makeFetch(() => (shouldFail ? statusResponse(500) : okResponse())), + }), + ); + + sink.track('evt'); + await sink.flush(); + expect( + readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')), + ).toHaveLength(1); + + shouldFail = false; + await sink.retryDiskEvents(); + expect( + readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')), + ).toHaveLength(0); + }); +}); diff --git a/packages/agent-core-v2/test/telemetry/consoleSink.test.ts b/packages/agent-core-v2/test/telemetry/consoleSink.test.ts new file mode 100644 index 000000000..1f65a4600 --- /dev/null +++ b/packages/agent-core-v2/test/telemetry/consoleSink.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from 'vitest'; + +import { ConsoleSink } from '#/telemetry/consoleSink'; + +describe('ConsoleSink', () => { + it('logs event name and properties with the default prefix', () => { + const lines: string[] = []; + const sink = new ConsoleSink({ log: (message) => lines.push(message) }); + sink.track('tool.call', { name: 'bash', count: 1 }); + expect(lines).toHaveLength(1); + expect(lines[0]).toContain('[telemetry] tool.call'); + expect(lines[0]).toContain('"name":"bash"'); + expect(lines[0]).toContain('"count":1'); + }); + + it('uses a custom prefix', () => { + const lines: string[] = []; + const sink = new ConsoleSink({ prefix: '[dbg]', log: (message) => lines.push(message) }); + sink.track('evt'); + expect(lines[0]).toBe('[dbg] evt'); + }); + + it('omits the payload when properties is undefined', () => { + const lines: string[] = []; + const sink = new ConsoleSink({ log: (message) => lines.push(message) }); + sink.track('evt'); + expect(lines[0]).toBe('[telemetry] evt'); + }); + + it('pretty-prints properties when requested', () => { + const lines: string[] = []; + const sink = new ConsoleSink({ pretty: true, log: (message) => lines.push(message) }); + sink.track('evt', { a: 1 }); + expect(lines[0]).toContain('\n'); + }); +}); diff --git a/packages/agent-core-v2/test/telemetry/telemetryService.test.ts b/packages/agent-core-v2/test/telemetry/telemetryService.test.ts index ac1114490..9a5062e1b 100644 --- a/packages/agent-core-v2/test/telemetry/telemetryService.test.ts +++ b/packages/agent-core-v2/test/telemetry/telemetryService.test.ts @@ -1,8 +1,12 @@ -import { beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, _clearScopedRegistryForTests, registerScopedService } from '#/_base/di/scope'; import { createScopedTestHost } from '#/_base/di/test'; +import { + resetUnexpectedErrorHandler, + setUnexpectedErrorHandler, +} from '#/_base/errors/unexpectedError'; import { type TelemetryClient, type TelemetryProperties, @@ -12,9 +16,17 @@ import { class CapturingClient implements TelemetryClient { readonly events: { event: string; properties?: TelemetryProperties }[] = []; + flushCalls = 0; + shutdownCalls = 0; track(event: string, properties?: TelemetryProperties): void { this.events.push({ event, properties }); } + flush(): void { + this.flushCalls += 1; + } + shutdown(): void { + this.shutdownCalls += 1; + } } describe('TelemetryService (unit)', () => { @@ -55,6 +67,127 @@ describe('TelemetryService (unit)', () => { svc.track('evt', { sessionId: 'override' }); expect(client.events[0]?.properties?.['sessionId']).toBe('override'); }); + + it('fans out to every sink passed via clients', () => { + const a = new CapturingClient(); + const b = new CapturingClient(); + const svc = new TelemetryService({ clients: [a, b] }); + svc.track('evt', { x: 1 }); + expect(a.events).toEqual([{ event: 'evt', properties: { x: 1 } }]); + expect(b.events).toEqual([{ event: 'evt', properties: { x: 1 } }]); + }); + + it('addSink registers a sink and its disposable removes it', () => { + const a = new CapturingClient(); + const b = new CapturingClient(); + const svc = new TelemetryService({ client: a }); + const disposable = svc.addSink(b); + svc.track('first'); + expect(a.events).toHaveLength(1); + expect(b.events).toHaveLength(1); + disposable.dispose(); + svc.track('second'); + expect(a.events).toHaveLength(2); + expect(b.events).toHaveLength(1); + }); + + it('removeSink stops delivery to that sink', () => { + const a = new CapturingClient(); + const b = new CapturingClient(); + const svc = new TelemetryService({ clients: [a, b] }); + svc.removeSink(a); + svc.track('evt'); + expect(a.events).toHaveLength(0); + expect(b.events).toHaveLength(1); + }); + + it('setEnabled(false) drops track; setEnabled(true) resumes', () => { + const client = new CapturingClient(); + const svc = new TelemetryService({ client }); + svc.setEnabled(false); + svc.track('dropped'); + expect(client.events).toHaveLength(0); + svc.setEnabled(true); + svc.track('sent'); + expect(client.events).toEqual([{ event: 'sent', properties: {} }]); + }); + + it('withContext child inherits enabled state at creation', () => { + const client = new CapturingClient(); + const root = new TelemetryService({ client }); + root.setEnabled(false); + const child = root.withContext({ turnId: 't1' }); + child.track('dropped'); + expect(client.events).toHaveLength(0); + }); + + it('flush fans out to every sink', async () => { + const a = new CapturingClient(); + const b = new CapturingClient(); + const svc = new TelemetryService({ clients: [a, b] }); + await svc.flush(); + expect(a.flushCalls).toBe(1); + expect(b.flushCalls).toBe(1); + }); + + it('shutdown fans out to every sink', async () => { + const a = new CapturingClient(); + const b = new CapturingClient(); + const svc = new TelemetryService({ clients: [a, b] }); + await svc.shutdown(); + expect(a.shutdownCalls).toBe(1); + expect(b.shutdownCalls).toBe(1); + }); + + it('flush is a no-op for sinks without flush', async () => { + const minimal: TelemetryClient = { track() {} }; + const svc = new TelemetryService({ client: minimal }); + await expect(svc.flush()).resolves.toBeUndefined(); + await expect(svc.shutdown()).resolves.toBeUndefined(); + }); +}); + +describe('TelemetryService (error isolation)', () => { + beforeEach(() => setUnexpectedErrorHandler(() => {})); + afterEach(() => resetUnexpectedErrorHandler()); + + it('a throwing sink does not prevent delivery to other sinks', () => { + const bad: TelemetryClient = { + track() { + throw new Error('boom'); + }, + }; + const good = new CapturingClient(); + const svc = new TelemetryService({ clients: [bad, good] }); + expect(() => svc.track('evt')).not.toThrow(); + expect(good.events).toEqual([{ event: 'evt', properties: {} }]); + }); + + it('flush tolerates a rejecting sink and still flushes the rest', async () => { + const bad: TelemetryClient = { + track() {}, + async flush() { + throw new Error('boom'); + }, + }; + const good = new CapturingClient(); + const svc = new TelemetryService({ clients: [bad, good] }); + await expect(svc.flush()).resolves.toBeUndefined(); + expect(good.flushCalls).toBe(1); + }); + + it('shutdown tolerates a rejecting sink and still shuts down the rest', async () => { + const bad: TelemetryClient = { + track() {}, + async shutdown() { + throw new Error('boom'); + }, + }; + const good = new CapturingClient(); + const svc = new TelemetryService({ clients: [bad, good] }); + await expect(svc.shutdown()).resolves.toBeUndefined(); + expect(good.shutdownCalls).toBe(1); + }); }); describe('ITelemetryService (scoped)', () => { diff --git a/packages/agent-core-v2/test/workspace/workspace.test.ts b/packages/agent-core-v2/test/workspace/workspace.test.ts index db6b1baa3..8f21e1d37 100644 --- a/packages/agent-core-v2/test/workspace/workspace.test.ts +++ b/packages/agent-core-v2/test/workspace/workspace.test.ts @@ -1,64 +1,131 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import type { Kaos } from '@moonshot-ai/kaos'; +import { beforeEach, describe, expect, it } from 'vitest'; -import { DisposableStore } from '#/_base/di/lifecycle'; -import { createServices } from '#/_base/di/test'; -import type { TestInstantiationService } from '#/_base/di/test'; -import { IKaosFactory } from '#/kaos/kaos'; -import { IWorkspaceFsService, IWorkspaceRegistry } from '#/workspace/workspace'; -import { WorkspaceFsService, WorkspaceRegistry } from '#/workspace/workspaceService'; -import { registerLogServices } from '../log/stubs'; +import { InstantiationType } from '#/_base/di/extensions'; +import { + LifecycleScope, + _clearScopedRegistryForTests, + registerScopedService, +} from '#/_base/di/scope'; +import { createScopedTestHost, stubPair } from '#/_base/di/test'; +import { PathSecurityError } from '#/_base/tools/policies/path-access'; +import { ISessionKaosService } from '#/kaos/kaos'; +import { ILogService } from '#/log/log'; +import { IWorkspaceService } from '#/workspace/workspace'; +import { WorkspaceService } from '#/workspace/workspaceService'; -describe('WorkspaceRegistry', () => { - let disposables: DisposableStore; - let ix: TestInstantiationService; +import { stubLog } from '../log/stubs'; +function fakeKaos(cwd: string): Kaos { + return { + pathClass: () => 'posix', + getcwd: () => cwd, + gethome: () => '/home/user', + } as unknown as Kaos; +} + +function stubSessionKaos(cwd: string, initialAdditional: readonly string[] = []): ISessionKaosService { + const kaos = fakeKaos(cwd); + const additional = [...initialAdditional]; + return { + _serviceBrand: undefined, + toolKaos: kaos, + persistenceKaos: kaos, + systemContextKaos: kaos, + get additionalDirs() { + return additional; + }, + setToolKaos: () => {}, + setPersistenceKaos: () => {}, + addAdditionalDir: (dir) => { + if (!additional.includes(dir)) additional.push(dir); + }, + removeAdditionalDir: (dir) => { + const idx = additional.indexOf(dir); + if (idx >= 0) additional.splice(idx, 1); + }, + }; +} + +describe('WorkspaceService', () => { beforeEach(() => { - disposables = new DisposableStore(); - ix = createServices(disposables, { - base: [registerLogServices], - additionalServices: (reg) => { - reg.definePartialInstance(IKaosFactory, {}); - reg.define(IWorkspaceRegistry, WorkspaceRegistry); - }, - }); + _clearScopedRegistryForTests(); + registerScopedService( + LifecycleScope.Session, + IWorkspaceService, + WorkspaceService, + InstantiationType.Delayed, + 'workspace', + ); }); - afterEach(() => disposables.dispose()); - it('register / get / list', () => { - const reg = ix.get(IWorkspaceRegistry); - const ws = reg.register('/repo'); - expect(ws.root).toBe('/repo'); - expect(reg.get(ws.id)).toEqual(ws); - expect(reg.list()).toEqual([ws]); - }); -}); - -describe('WorkspaceFsService', () => { - let disposables: DisposableStore; - let ix: TestInstantiationService; - - beforeEach(() => { - disposables = new DisposableStore(); - ix = createServices(disposables, { - base: [registerLogServices], - additionalServices: (reg) => { - reg.definePartialInstance(IKaosFactory, {}); - reg.define(IWorkspaceRegistry, WorkspaceRegistry); - reg.define(IWorkspaceFsService, WorkspaceFsService); - }, - }); - }); - afterEach(() => disposables.dispose()); - - it('resolves a relative path against a registered workspace', () => { - const reg = ix.get(IWorkspaceRegistry); - const ws = reg.register('/repo'); - const fs = ix.createInstance(WorkspaceFsService, reg); - expect(fs.resolve(ws.id, 'src/index.ts')).toBe('/repo/src/index.ts'); - }); - - it('throws for unknown workspace', () => { - const fs = ix.get(IWorkspaceFsService); - expect(() => fs.resolve('nope', 'x')).toThrow(/unknown workspace/); + function build(sessionKaos: ISessionKaosService): IWorkspaceService { + const host = createScopedTestHost([stubPair(ILogService, stubLog())]); + const session = host.child(LifecycleScope.Session, 's1', [stubPair(ISessionKaosService, sessionKaos)]); + return session.accessor.get(IWorkspaceService); + } + + it('reflects the session kaos cwd and additional dirs', () => { + const ws = build(stubSessionKaos('/repo', ['/extra'])); + expect(ws.workDir).toBe('/repo'); + expect(ws.additionalDirs).toEqual(['/extra']); + }); + + it('resolves a relative path against workDir', () => { + const ws = build(stubSessionKaos('/repo')); + expect(ws.resolve('src/index.ts')).toBe('/repo/src/index.ts'); + }); + + it('normalizes an absolute path', () => { + const ws = build(stubSessionKaos('/repo')); + expect(ws.resolve('/repo/a/../b')).toBe('/repo/b'); + }); + + it('checks whether a path is within the workspace', () => { + const ws = build(stubSessionKaos('/repo', ['/extra'])); + expect(ws.isWithin('src/index.ts')).toBe(true); + expect(ws.isWithin('/repo/sub/file')).toBe(true); + expect(ws.isWithin('/extra/file')).toBe(true); + expect(ws.isWithin('/elsewhere/file')).toBe(false); + }); + + it('allows a path inside the workspace', () => { + const ws = build(stubSessionKaos('/repo')); + expect(ws.assertAllowed('src/index.ts', 'read')).toBe('/repo/src/index.ts'); + }); + + it('rejects a relative path that escapes the workspace', () => { + const ws = build(stubSessionKaos('/repo')); + expect(() => ws.assertAllowed('../outside', 'read')).toThrowError(PathSecurityError); + try { + ws.assertAllowed('../outside', 'read'); + } catch (error) { + expect((error as PathSecurityError).code).toBe('PATH_OUTSIDE_WORKSPACE'); + } + }); + + it('rejects a sensitive file', () => { + const ws = build(stubSessionKaos('/repo')); + expect(() => ws.assertAllowed('.env', 'read')).toThrowError(PathSecurityError); + try { + ws.assertAllowed('.env', 'read'); + } catch (error) { + expect((error as PathSecurityError).code).toBe('PATH_SENSITIVE'); + } + }); + + it('snapshots a WorkspaceConfig value object', () => { + const ws = build(stubSessionKaos('/repo', ['/extra'])); + expect(ws.toConfig()).toEqual({ workspaceDir: '/repo', additionalDirs: ['/extra'] }); + }); + + it('delegates additional-dir mutation to the session kaos', () => { + const ws = build(stubSessionKaos('/repo')); + ws.addAdditionalDir('/extra'); + expect(ws.additionalDirs).toEqual(['/extra']); + ws.addAdditionalDir('/extra'); // idempotent + expect(ws.additionalDirs).toEqual(['/extra']); + ws.removeAdditionalDir('/extra'); + expect(ws.additionalDirs).toEqual([]); }); });