feat(agent-core-v2): flesh out auth, config, telemetry and workspace services

- auth: device-code OAuth flow (startLogin/getFlow/cancelLogin),
  resolveTokenProvider, provider config schemas, ensureReady,
  AUTH_LOGIN_REQUIRED error code
- config: schema-driven registry (validate/merge/defaultValue/diagnostics),
  reload/getAll/ready, new Session-scoped ISessionConfigService replacing
  IAgentConfigService, pure helpers in configPure
- telemetry: multi-sink fan-out with flush/shutdown, add ConsoleSink,
  CloudSink and CloudTransport
- workspace: Session-scoped IWorkspaceService (workDir/additionalDirs,
  resolve/isWithin/assertAllowed) replacing the registry/fs split
- kaos: add cwd/chdir; gateway: add flushLogs/flushGlobalLogs
- docs: add Permission design doc, refresh service-design and
  di-scope-domains for ISessionConfigService
- update kosong/tool/gateway consumers and tests to ISessionConfigService
This commit is contained in:
haozhe.yang 2026-06-26 00:32:38 +08:00
parent f6c2c513dc
commit 3122e0803a
42 changed files with 2816 additions and 386 deletions

View file

@ -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<ExecutableToolResult>;
}
```
`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 自注册**:拥有专属维度的 domainplan/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 个节点、首个命中短路,远未触及。

View file

@ -86,7 +86,7 @@ package "Turn scope (per turn)" #FDEDEC {
}
package "Multi-scope" #F5EEF8 {
rectangle "<b>config</b>\n<size:9><i>Multi</i></size>\n IConfigRegistry(C)\n IConfigService(C)\n IAgentConfigService(A)" as config #E8DAEF
rectangle "<b>config</b>\n<size:9><i>Multi</i></size>\n IConfigRegistry(C)\n IConfigService(C)\n ISessionConfigService(S)" as config #E8DAEF
rectangle "<b>kosong</b>\n<size:9><i>Multi</i></size>\n IModelCatalogService(C)\n IProviderManager(S)\n ILLMService(A)" as kosong #E8DAEF
rectangle "<b>records</b>\n<size:9><i>Multi</i></size>\n ISessionStore(C)\n ISessionMetaStore(S)\n IAgentRecords(A)" as records #E8DAEF
rectangle "<b>tool</b>\n<size:9><i>Multi</i></size>\n IToolDefinitionRegistry(C)\n IToolService(A)" as tool #E8DAEF

File diff suppressed because one or more lines are too long

Before

Width:  |  Height:  |  Size: 197 KiB

After

Width:  |  Height:  |  Size: 197 KiB

Before After
Before After

View file

@ -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`

View file

@ -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')

View file

@ -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<ErrorCode, ErrorInfo>;
export function errorInfo(code: ErrorCode): ErrorInfo {

View file

@ -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<void>;
logout(provider: string): Promise<void>;
status(): Promise<AuthStatus>;
startLogin(provider?: string): Promise<OAuthFlowStart>;
getFlow(provider?: string): OAuthFlowSnapshot | undefined;
cancelLogin(provider?: string): Promise<OAuthLoginCancelResponse>;
logout(provider?: string): Promise<OAuthLogoutResponse>;
status(provider?: string): Promise<AuthStatus>;
resolveTokenProvider(provider: string, oauthRef?: OAuthRef): BearerTokenProvider | undefined;
}
export const IOAuthService: ServiceIdentifier<IOAuthService> =
@ -27,6 +41,7 @@ export const IOAuthService: ServiceIdentifier<IOAuthService> =
export interface IAuthSummaryService {
readonly _serviceBrand: undefined;
summarize(): Promise<readonly AuthStatus[]>;
ensureReady(provider?: string): Promise<void>;
}
export const IAuthSummaryService: ServiceIdentifier<IAuthSummaryService> =

View file

@ -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<typeof setTimeout> | undefined;
errorMessage: string | undefined;
resolvedAt: string | undefined;
}
export class OAuthService extends Disposable implements IOAuthService {
declare readonly _serviceBrand: undefined;
private readonly loggedIn = new Set<string>();
private readonly toolkit: KimiOAuthToolkit;
private readonly flows = new Map<string, FlowState>();
constructor(
@IConfigService _config: IConfigService,
@IEnvironmentService _env: IEnvironmentService,
@ITelemetryService _telemetry: ITelemetryService,
) {}
login(provider: string): Promise<void> {
this.loggedIn.add(provider);
return Promise.resolve();
}
logout(provider: string): Promise<void> {
this.loggedIn.delete(provider);
return Promise.resolve();
}
status(): Promise<AuthStatus> {
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<OAuthFlowStart> {
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<DeviceAuthorization>((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<OAuthLoginCancelResponse> {
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<OAuthLogoutResponse> {
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<AuthStatus> {
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<ProvidersSection>(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<ProvidersSection>(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<readonly AuthStatus[]> {
if (this.oauth === undefined) return Promise.resolve([]);
return this.oauth.status().then((s) => [s]);
async summarize(): Promise<readonly AuthStatus[]> {
const providers = this.config.get<ProvidersSection>(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<void> {
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');

View file

@ -6,3 +6,4 @@
export * from './auth';
export * from './authService';
export * from './oauthSchemas';

View file

@ -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<typeof ProviderTypeSchema>;
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<typeof OAuthRefSchema>;
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<typeof ProviderConfigSchema>;
export const PROVIDERS_SECTION = 'providers';
export const ProvidersSectionSchema = z.record(z.string(), ProviderConfigSchema);
export type ProvidersSection = z.infer<typeof ProvidersSectionSchema>;

View file

@ -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<T> {
parse(value: unknown): T;
}
export type ConfigMerge<T> = (base: T | undefined, patch: unknown) => T;
export interface ConfigSection<T = unknown> {
readonly domain: string;
readonly schema: unknown;
readonly schema?: ConfigSchema<T>;
readonly defaultValue?: T;
readonly merge: ConfigMerge<T>;
}
export interface RegisterSectionOptions<T> {
readonly defaultValue?: T;
readonly merge?: ConfigMerge<T>;
}
export interface IConfigRegistry {
readonly _serviceBrand: undefined;
registerSection(domain: string, schema: unknown): void;
registerSection<T>(domain: string, schema: ConfigSchema<T>, options?: RegisterSectionOptions<T>): void;
getSection(domain: string): ConfigSection | undefined;
merge(base: unknown, patch: unknown): unknown;
listSections(): readonly ConfigSection[];
validate<T>(domain: string, value: unknown): T;
merge<T>(domain: string, base: T | undefined, patch: unknown): T;
defaultValue<T>(domain: string): T | undefined;
}
export const IConfigRegistry: ServiceIdentifier<IConfigRegistry> =
createDecorator<IConfigRegistry>('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<string, unknown>;
export interface IConfigService {
readonly _serviceBrand: undefined;
readonly ready: Promise<void>;
readonly onDidChange: Event<ConfigChangedEvent>;
get<T = unknown>(domain: string): T;
getAll(): ResolvedConfig;
set(domain: string, patch: unknown): Promise<void>;
reload(): Promise<void>;
diagnostics(): readonly ConfigDiagnostic[];
}
export const IConfigService: ServiceIdentifier<IConfigService> =
createDecorator<IConfigService>('configService');
export interface IAgentConfigService {
export interface SessionConfigSection {
readonly modelAlias?: string;
readonly thinkingLevel?: string;
readonly systemPrompt?: string;
readonly provider?: string;
}
export type SessionConfigPatch = Partial<
Pick<SessionConfigSection, 'modelAlias' | 'thinkingLevel' | 'systemPrompt'>
>;
export interface SessionConfigChangedEvent {
readonly changed: readonly (keyof SessionConfigSection)[];
}
export interface ISessionConfigService {
readonly _serviceBrand: undefined;
readonly ready: Promise<void>;
readonly onDidChange: Event<SessionConfigChangedEvent>;
readonly modelAlias: string | undefined;
readonly thinkingLevel: string | undefined;
readonly systemPrompt: string | undefined;
readonly provider: string | undefined;
readonly cwd: string;
update(patch: SessionConfigPatch): Promise<void>;
setModel(alias: string): Promise<void>;
setThinking(level: string): Promise<void>;
}
export const IAgentConfigService: ServiceIdentifier<IAgentConfigService> =
createDecorator<IAgentConfigService>('agentConfigService');
export const ISessionConfigService: ServiceIdentifier<ISessionConfigService> =
createDecorator<ISessionConfigService>('sessionConfigService');

View file

@ -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<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value);
}
export function deepMerge<T>(base: T | undefined, patch: unknown): T {
if (!isPlainObject(base) || !isPlainObject(patch)) {
return (patch ?? base) as T;
}
const out: Record<string, unknown> = { ...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<T extends Record<string, unknown>>(value: T): Partial<T> {
const out: Partial<T> = {};
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);
}

View file

@ -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<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value);
}
function deepMerge<T>(base: T, patch: unknown): T {
if (!isPlainObject(base) || !isPlainObject(patch)) {
return (patch ?? base) as T;
}
const out: Record<string, unknown> = { ...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<string, unknown>();
private readonly sections = new Map<string, ConfigSection>();
registerSection(domain: string, schema: unknown): void {
registerSection<T>(
domain: string,
schema: ConfigSchema<T>,
options: RegisterSectionOptions<T> = {},
): 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<unknown>,
defaultValue: options.defaultValue,
merge: (options.merge ?? deepMerge) as ConfigMerge<unknown>,
});
}
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<T>(domain: string, value: unknown): T {
const schema = this.sections.get(domain)?.schema;
return (schema === undefined ? value : schema.parse(value)) as T;
}
merge<T>(domain: string, base: T | undefined, patch: unknown): T {
const merge = this.sections.get(domain)?.merge ?? deepMerge;
return merge(base, patch) as T;
}
defaultValue<T>(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<ConfigChangedEvent>());
readonly onDidChange: Event<ConfigChangedEvent> = this._onDidChange.event;
private readonly root = new Map<string, unknown>();
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<T = unknown>(domain: string): T {
return this.root.get(domain) as T;
return this.effective[domain] as T;
}
set(domain: string, patch: unknown): Promise<void> {
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<void> {
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<void> {
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<AgentSection>('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<void> {
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<void> {
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');

View file

@ -6,3 +6,4 @@
export * from './config';
export * from './configService';
export * from './sessionConfigService';

View file

@ -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<SessionConfigChangedEvent>());
readonly onDidChange: Event<SessionConfigChangedEvent> = this._onDidChange.event;
readonly ready: Promise<void>;
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<SessionConfigSection>('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<void> {
const clean = omitUndefined(patch as Record<string, unknown>) as SessionConfigPatch;
if (Object.keys(clean).length === 0) return;
await this.meta.write(clean);
this.apply(clean, true);
}
setModel(alias: string): Promise<void> {
return this.update({ modelAlias: alias });
}
setThinking(level: string): Promise<void> {
return this.update({ thinkingLevel: level });
}
private async restore(): Promise<void> {
try {
const stored = await this.meta.read();
this.apply(stored as Partial<SessionConfigSection>, false);
} catch (error) {
this.log.warn('session config restore failed', { error: describeUnknownError(error) });
}
}
private apply(patch: Partial<SessionConfigSection>, 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',
);

View file

@ -31,6 +31,8 @@ export interface IRestGateway {
steer(sessionId: string, agentId: string, content: string): Promise<void>;
cancel(sessionId: string, agentId: string, reason?: string): Promise<void>;
getStatus(sessionId: string): Promise<unknown>;
flushLogs(sessionId: string): Promise<void>;
flushGlobalLogs(): Promise<void>;
}
export const IRestGateway: ServiceIdentifier<IRestGateway> =

View file

@ -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<unknown> {
return Promise.resolve(this.scopes.get(sessionId) !== undefined);
}
async flushLogs(sessionId: string): Promise<void> {
const session = this.scopes.get(sessionId);
if (session === undefined) return;
await session.accessor.get(ISessionLogService).flush();
}
flushGlobalLogs(): Promise<void> {
return this.log.flush();
}
}
export class WSGateway implements IWSGateway {

View file

@ -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<void>;
}
// eslint-disable-next-line @typescript-eslint/no-redeclare

View file

@ -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<void> {
return Promise.reject(new Error('KaosService.chdir is not supported'));
}
}
registerScopedService(

View file

@ -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<GenerateResult> {
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})`);
}

View file

@ -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,

View file

@ -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<string | null>;
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<void>;
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<typeof setInterval> | 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<void> {
if (this.buffer.length === 0) return;
const events = this.buffer;
this.buffer = [];
await this.transport.send(events);
}
async shutdown(): Promise<void> {
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<void> {
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;
}

View file

@ -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<string, CloudPrimitive>;
export type CloudContext = Record<string, CloudPrimitive>;
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<string, CloudPrimitive>[];
}
export interface CloudTransportOptions {
readonly homeDir: string;
readonly deviceId: string;
readonly endpoint?: string;
readonly getAccessToken?: () => string | null | Promise<string | null>;
readonly fetchImpl?: typeof fetch;
readonly retryBackoffsMs?: readonly number[];
readonly requestTimeoutMs?: number;
readonly sleep?: (ms: number, signal?: AbortSignal) => Promise<void>;
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<string | null>) | null;
private readonly fetchImpl: typeof fetch;
private readonly retryBackoffsMs: readonly number[];
private readonly requestTimeoutMs: number;
private readonly sleepImpl: (ms: number, signal?: AbortSignal) => Promise<void>;
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<void> {
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<void> {
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<void> {
const token = this.getAccessToken === null ? null : await this.getAccessToken();
const headers: Record<string, string> = {
'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<string, string>,
signal?: AbortSignal,
): Promise<Response> {
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<string, CloudPrimitive> {
const out: Record<string, CloudPrimitive> = {};
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<string, CloudPrimitive>, 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<Response> {
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<void> {
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');
}

View file

@ -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);
}

View file

@ -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';

View file

@ -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> | void;
shutdown?(): Promise<void> | 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<void>;
shutdown(): Promise<void>;
}
export const noopTelemetryClient: TelemetryClient = {
track: () => {},
withContext: () => noopTelemetryClient,
setContext: () => {},
flush: () => {},
shutdown: () => {},
};
// eslint-disable-next-line @typescript-eslint/no-redeclare

View file

@ -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<void> {
await Promise.all(
this.sinks.map((sink) =>
Promise.resolve(sink.flush?.()).catch(onUnexpectedError),
),
);
}
async shutdown(): Promise<void> {
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<string, Exclude<TelemetryProperties[string], undefined>> = {};
for (const [key, value] of Object.entries(input)) {

View file

@ -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,

View file

@ -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';

View file

@ -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';

View file

@ -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<IWorkspaceRegistry> =
createDecorator<IWorkspaceRegistry>('workspaceRegistry');
export interface IWorkspaceFsService {
readonly _serviceBrand: undefined;
resolve(workspaceId: string, rel: string): string;
}
export const IWorkspaceFsService: ServiceIdentifier<IWorkspaceFsService> =
createDecorator<IWorkspaceFsService>('workspaceFsService');
export const IWorkspaceService: ServiceIdentifier<IWorkspaceService> =
createDecorator<IWorkspaceService>('workspaceService');

View file

@ -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<string, WorkspaceInfo>();
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',
);

View file

@ -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<void> => new Promise((resolve) => setTimeout(resolve, 0));
interface FakeToolkit {
readonly login: ReturnType<typeof vi.fn>;
readonly logout: ReturnType<typeof vi.fn>;
readonly getCachedAccessToken: ReturnType<typeof vi.fn>;
readonly tokenProvider: ReturnType<typeof vi.fn>;
}
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<typeof vi.fn>;
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();
});
});

View file

@ -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<string, unknown> = {}) {
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<string, unknown>;
const agentKaos: IKaosService = {
_serviceBrand: undefined,
get kaos(): never {
throw new Error('unused');
},
cwd: '/repo',
chdir: () => Promise.resolve(),
};
let sessionSection: SessionConfigSection;
let metaStore: ReturnType<typeof stubSessionMetaStore>;
beforeEach(() => {
disposables = new DisposableStore();
agentSection = {};
sessionSection = {};
metaStore = stubSessionMetaStore();
ix = createServices(disposables, {
base: [registerRecordsServices],
base: [registerLogServices],
additionalServices: (reg) => {
reg.definePartialInstance(IConfigService, { get: <T>() => agentSection as T });
reg.defineInstance(IKaosService, agentKaos);
reg.define(IAgentConfigService, AgentConfigService);
reg.definePartialInstance(IConfigService, { get: <T>() => 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: <T>() => 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']);
});
});

View file

@ -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, {});
}

View file

@ -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));
}

View file

@ -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<void> };
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();
});
});

View file

@ -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);

View file

@ -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<string, string>;
readonly body: {
readonly user_id: string;
readonly events: readonly Record<string, unknown>[];
};
}
type Responder = (req: CapturedRequest) => Response | Promise<Response>;
function makeFetch(responder: Responder): typeof fetch {
return (async (input: unknown, init: unknown) => {
const requestInit = init as { headers: Record<string, string>; 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> = {}): 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);
});
});

View file

@ -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');
});
});

View file

@ -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)', () => {

View file

@ -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([]);
});
});