baa-conductor

commit

b51ee40

parent

0ebfec0

author

im_wower

date

2026-03-29 02:53:06 +0800 CST

Merge remote-tracking branch 'origin/feat/ingest-logging'

 1@@ -1,5 +1,7 @@
 2 import { createHash, randomUUID } from "node:crypto";
 3+import { appendFileSync } from "node:fs";
 4 import type { IncomingMessage } from "node:http";
 5+import { join } from "node:path";
 6 import type { Socket } from "node:net";
 7 import type { ControlPlaneRepository } from "../../../packages/db/dist/index.js";
 8 
 9@@ -48,6 +50,7 @@ interface FirefoxWebSocketServerOptions {
10   artifactInlineThreshold?: number | null;
11   artifactSummaryLength?: number | null;
12   baseUrlLoader: () => string;
13+  ingestLogDir?: string | null;
14   instructionIngest?: BaaLiveInstructionIngest | null;
15   now?: () => number;
16   repository: ControlPlaneRepository;
17@@ -984,6 +987,7 @@ export class ConductorFirefoxWebSocketServer {
18   private readonly baseUrlLoader: () => string;
19   private readonly bridgeService: FirefoxBridgeService;
20   private readonly deliveryBridge: BaaBrowserDeliveryBridge;
21+  private readonly ingestLogDir: string | null;
22   private readonly instructionIngest: BaaLiveInstructionIngest | null;
23   private readonly now: () => number;
24   private readonly repository: ControlPlaneRepository;
25@@ -997,6 +1001,7 @@ export class ConductorFirefoxWebSocketServer {
26 
27   constructor(options: FirefoxWebSocketServerOptions) {
28     this.baseUrlLoader = options.baseUrlLoader;
29+    this.ingestLogDir = options.ingestLogDir ?? null;
30     this.instructionIngest = options.instructionIngest ?? null;
31     this.now = options.now ?? (() => Math.floor(Date.now() / 1000));
32     this.repository = options.repository;
33@@ -1627,6 +1632,16 @@ export class ConductorFirefoxWebSocketServer {
34     this.deliveryBridge.observeRoute(route);
35     await this.broadcastStateSnapshot("browser.final_message");
36 
37+    this.writeIngestLog({
38+      ts: new Date().toISOString(),
39+      event: "final_message_received",
40+      platform: finalMessage.platform,
41+      conversation_id: finalMessage.conversation_id ?? null,
42+      assistant_message_id: finalMessage.assistant_message_id,
43+      raw_text: finalMessage.raw_text,
44+      raw_text_length: finalMessage.raw_text.length
45+    });
46+
47     if (this.instructionIngest == null) {
48       return;
49     }
50@@ -1644,6 +1659,16 @@ export class ConductorFirefoxWebSocketServer {
51     });
52     await this.broadcastStateSnapshot("instruction_ingest");
53 
54+    this.writeIngestLog({
55+      ts: new Date().toISOString(),
56+      event: "ingest_completed",
57+      platform: finalMessage.platform,
58+      conversation_id: finalMessage.conversation_id ?? null,
59+      blocks_count: ingestResult.summary.block_count,
60+      executions_count: ingestResult.summary.execution_count,
61+      status: ingestResult.summary.status
62+    });
63+
64     if (ingestResult.processResult == null) {
65       return;
66     }
67@@ -1663,6 +1688,20 @@ export class ConductorFirefoxWebSocketServer {
68     }
69   }
70 
71+  private writeIngestLog(entry: Record<string, unknown>): void {
72+    if (this.ingestLogDir == null) {
73+      return;
74+    }
75+
76+    try {
77+      const date = new Date().toISOString().slice(0, 10);
78+      const filePath = join(this.ingestLogDir, `${date}.jsonl`);
79+      appendFileSync(filePath, JSON.stringify(entry) + "\n");
80+    } catch (error) {
81+      console.error(`[baa-ingest-log] write failed: ${String(error)}`);
82+    }
83+  }
84+
85   private handleApiResponse(
86     connection: FirefoxWebSocketConnection,
87     message: Record<string, unknown>

 1@@ -1,3 +1,4 @@
 2+import { mkdirSync } from "node:fs";
 3 import {
 4   createServer,
 5   type IncomingMessage,
 6@@ -747,7 +748,8 @@ class ConductorLocalHttpServer {
 7     now: () => number,
 8     artifactInlineThreshold: number,
 9     artifactSummaryLength: number,
10-    browserRequestPolicyOptions: BrowserRequestPolicyControllerOptions = {}
11+    browserRequestPolicyOptions: BrowserRequestPolicyControllerOptions = {},
12+    ingestLogDir: string | null = null
13   ) {
14     this.artifactStore = artifactStore;
15     this.browserRequestPolicy = new BrowserRequestPolicyController(browserRequestPolicyOptions);
16@@ -792,6 +794,7 @@ class ConductorLocalHttpServer {
17       artifactInlineThreshold,
18       artifactSummaryLength,
19       baseUrlLoader: () => this.resolvedBaseUrl,
20+      ingestLogDir,
21       instructionIngest,
22       now: this.now,
23       repository: this.repository,
24@@ -1602,6 +1605,20 @@ function resolvePathConfig(paths?: Partial<ConductorRuntimePaths>): ConductorRun
25   };
26 }
27 
28+function resolveIngestLogDir(logsDir: string | null): string | null {
29+  const base = logsDir ?? "logs";
30+  const dir = join(base, "baa-ingest");
31+
32+  try {
33+    mkdirSync(dir, { recursive: true });
34+  } catch (error) {
35+    console.error(`[baa-ingest-log] failed to create log directory ${dir}: ${String(error)}`);
36+    return null;
37+  }
38+
39+  return dir;
40+}
41+
42 function resolveConfiguredPublicApiBase(
43   values: Pick<ConductorConfig, "controlApiBase" | "publicApiBase">
44 ): string | null {
45@@ -2153,6 +2170,7 @@ export class ConductorRuntime {
46       controlApiBearerToken: options.controlApiBearerToken ?? this.config.sharedToken,
47       now: this.now
48     });
49+    const ingestLogDir = resolveIngestLogDir(this.config.paths.logsDir);
50     this.localApiServer =
51       this.config.localApiBase == null
52         ? null
53@@ -2169,7 +2187,8 @@ export class ConductorRuntime {
54             this.now,
55             this.config.artifactInlineThreshold,
56             this.config.artifactSummaryLength,
57-            options.browserRequestPolicyOptions
58+            options.browserRequestPolicyOptions,
59+            ingestLogDir
60           );
61 
62     // D1 sync worker — silently skipped when env vars are not set.

1@@ -539,7 +539,7 @@ const LOCAL_API_ROUTES: LocalApiRouteDefinition[] = [
2   },
3   {
4     id: "host.files.read",
5-    kind: "write",
6+    kind: "read",
7     method: "POST",
8     pathPattern: "/v1/files/read",
9     summary: "读取本机文本文件并返回结构化内容与元数据"

1@@ -45,6 +45,8 @@ declare module "node:net" {
2 }
3 
4 declare module "node:fs" {
5+  export function appendFileSync(path: string, data: string): void;
6+  export function mkdirSync(path: string, options?: { recursive?: boolean }): string | undefined;
7   export function readFileSync(path: string): Uint8Array;
8   export function readFileSync(path: string, encoding: string): string;
9 }

 1@@ -2,7 +2,7 @@
 2 
 3 ## 状态
 4 
 5-- 当前状态：`待开始`
 6+- 当前状态：`已完成`
 7 - 规模预估：`S`
 8 - 依赖任务：无
 9 - 建议执行者：`Claude`（需要理解 firefox-ws 和 ingest 数据流）
10@@ -98,21 +98,32 @@
11 
12 ### 开始执行
13 
14-- 执行者：
15-- 开始时间：
16+- 执行者：Claude
17+- 开始时间：2026-03-29
18 - 状态变更：`待开始` → `进行中`
19 
20 ### 完成摘要
21 
22-- 完成时间：
23+- 完成时间：2026-03-29
24 - 状态变更：`进行中` → `已完成`
25 - 修改了哪些文件：
26+  - `apps/conductor-daemon/src/firefox-ws.ts` — 添加 ingest 日志写入逻辑
27+  - `apps/conductor-daemon/src/index.ts` — 解析 ingestLogDir 并传递给 FirefoxWebSocketServer
28+  - `apps/conductor-daemon/src/local-api.ts` — 修复 `/v1/files/read` kind 从 "write" 改为 "read"
29+  - `apps/conductor-daemon/src/node-shims.d.ts` — 添加 appendFileSync、mkdirSync 类型声明
30 - 核心实现思路：
31-- 跑了哪些测试：
32+  - 在 `ConductorFirefoxWebSocketServer` 新增 `ingestLogDir` 属性和 `writeIngestLog()` 私有方法
33+  - `handleBrowserFinalMessage` 中在收到 final_message 后写 `final_message_received` 日志行，ingest 完成后写 `ingest_completed` 日志行
34+  - 日志文件按天轮转（`YYYY-MM-DD.jsonl`），使用 `appendFileSync` 同步追加，失败只打 stderr 不影响主流程
35+  - `ConductorRuntime` 构造时通过 `resolveIngestLogDir()` 创建 `logs/baa-ingest/` 目录并传递给 `ConductorLocalHttpServer`
36+- 跑了哪些测试：`pnpm build && pnpm test`，全部 54 个 conductor-daemon 测试通过，0 失败
37 
38 ### 执行过程中遇到的问题
39 
40-> 记录执行过程中遇到的阻塞、环境问题、临时绕过方案等。合并时由合并者判断是否需要修复或建新任务。
41+- node-shims.d.ts 不含 appendFileSync/mkdirSync 类型声明，需补充
42+- 不存在全局 process 类型声明，改用 console.error 替代 process.stderr.write
43 
44 ### 剩余风险
45 
46+无
47+