fix: 主线 A 事件落库地基 + 主线 B1/B2 执行原语

主线 A（事件驱动落库）: - 统一 EventStore 模块单例：RuntimeApp 不再 new EventStore，改用 eventStore 并 setRepositories(14 个 domain repo),消除事件流向空 DB 的割裂 - Scheduler.create_tasks 改为 async,真正发出 task.created 事件 - run.ts dispatchTask 加 await - 主线 A 独立复审发现并修复关键假绿：四个 repo（Task/Agent/ToolRun/ TaskAttempt）的 *Update 类型 Omit<'status'> 且 update() 主动丢弃 status, 导致 EventStore.project() 的状态写入全部静默失效,DB 行内容 tasks.status 永远冻结在 pending,UI 显示的 completed 来自内存 graph。已修,DB 现真实反映 task.status=completed - 补 agent.started/agent.completed/agent.failed 事件发出（之前 agents 表恒空),修复后 agents 表有正确行+status 主线 B1（结构化工具调用块类型,N1）: - 新增 content-block.ts 定义 Anthropic canonical content blocks (TextBlock/ThinkingBlock/ToolUseBlock/ToolResultBlock/CanonicalMessage) - provider.ts ProviderCompletionInput 去掉 unknown 逃生舱: messages: CanonicalMessage[], tools?: ToolDefinitionBlock[], tool_choice?: ToolChoice, system?: string | TextBlock[] 主线 B2（read-before-edit 代码层强制,FR-009）: - fs/index.ts 新增 readFileState 机制(移植 claude-code FileEditTool), fs.edit 执行前检查:未读先改报 "File has not been read yet",外部修改报 "File has been unexpectedly modified" - 修复 fs.edit 参数名不匹配:兼容 old_str/new_str (ExecutorRole) 和 find/replace (UI) 两种命名 - fs_edit 唯一性检查(非 global 模式下 old_str 出现多次报错) 真实验收: - TSC=0 - air run 后 DB:events=5(原 3,+agent.started/completed), tasks.status=completed(原 frozen pending),agents 1 行 status=completed - read-before-edit 行为测试:未读先改 status=error,读后再改 status=ok Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-08 12:10:07 +08:00
parent ddefcbb2b1
commit bac285d412
12 changed files with 315 additions and 37 deletions
--- a/packages/runtime/src/app/RuntimeApp.ts
+++ b/packages/runtime/src/app/RuntimeApp.ts
@@ -26,6 +26,17 @@ import { EventIngestorImpl } from '../events/EventIngestor.js'
 import { TaskRepository } from '../storage/repositories/TaskRepository.js'
 import { MessageRepository } from '../storage/repositories/MessageRepository.js'
 import { EvidenceRepository } from '../storage/repositories/EvidenceRepository.js'
+import { SessionRepository } from '../storage/repositories/SessionRepository.js'
+import { MessageDraftRepository } from '../storage/repositories/MessageDraftRepository.js'
+import { TaskAttemptRepository } from '../storage/repositories/TaskAttemptRepository.js'
+import { TaskDependencyRepository } from '../storage/repositories/TaskDependencyRepository.js'
+import { AgentRepository } from '../storage/repositories/AgentRepository.js'
+import { ToolRunRepository } from '../storage/repositories/ToolRunRepository.js'
+import { CommandRunRepository } from '../storage/repositories/CommandRunRepository.js'
+import { ArtifactRepository } from '../storage/repositories/ArtifactRepository.js'
+import { DiagnosticRepository } from '../storage/repositories/DiagnosticRepository.js'
+import { WorkspaceRepository } from '../storage/repositories/WorkspaceRepository.js'
+import { SummaryRepository } from '../storage/repositories/SummaryRepository.js'
 import { createCapabilityRegistry, type CapabilityRegistry } from '../capabilities/CapabilityRegistry.js'

 export interface RuntimeAppConfig {
@@ -79,9 +90,8 @@ export class RuntimeApp {
    const raw_db = this.db.getRawDatabase()
    // Wire singleton eventStore with real DB (EventIngestor uses it)
    if (raw_db) eventStore.setTransactionManager(this.db)
-    this.event_store = raw_db
-      ? new EventStore({ id: 'runtime', db: raw_db } as any)
-      : new EventStore({ id: 'startup', db: null } as any)
+    // Use module singleton eventStore - don't create separate instance
+    this.event_store = eventStore
    this.event_ingestor = new EventIngestorImpl()

    // Wire ProjectionStore → ProjectionClient (DD §13.2)
@@ -146,15 +156,34 @@ export class RuntimeApp {
    // Step 5: Hydrate ProjectionStore from DB (INV-5: from SQLite, not EventBus)
    this.logger.info('Hydrating projection store', { session_id: this.config.session_id })

-    // Step 6: Recover interrupted tasks (INV-5: rebuild from SQLite)
+    // Step 6: Wire all domain repositories to module singleton EventStore
    try {
      const raw_db = this.db.getRawDatabase()
      if (raw_db) {
-        const task_repo = new TaskRepository(raw_db as any)
-        const message_repo = new MessageRepository(raw_db as any)
-        const evidence_repo = new EvidenceRepository(raw_db as any)
-        this.context_assembler.set_data_sources({ message_repo, evidence_store: evidence_repo })
-        this.scheduler.set_task_repo(task_repo)
+        const sessionRepo = new SessionRepository(raw_db as any)
+        const messageRepo = new MessageRepository(raw_db as any)
+        const messageDraftRepo = new MessageDraftRepository(raw_db as any)
+        const taskRepo = new TaskRepository(raw_db as any)
+        const taskAttemptRepo = new TaskAttemptRepository(raw_db as any)
+        const taskDepRepo = new TaskDependencyRepository(raw_db as any)
+        const agentRepo = new AgentRepository(raw_db as any)
+        const toolRunRepo = new ToolRunRepository(raw_db as any)
+        const commandRunRepo = new CommandRunRepository(raw_db as any)
+        const artifactRepo = new ArtifactRepository(raw_db as any)
+        const diagnosticRepo = new DiagnosticRepository(raw_db as any)
+        const evidenceRepo = new EvidenceRepository(raw_db as any)
+        const workspaceRepo = new WorkspaceRepository(raw_db as any)
+        const summaryRepo = new SummaryRepository(raw_db as any)
+
+        this.event_store.setRepositories({
+          sessionRepo, messageRepo, messageDraftRepo, taskRepo, taskAttemptRepo,
+          taskDepRepo, agentRepo, toolRunRepo, commandRunRepo, artifactRepo,
+          diagnosticRepo, evidenceRepo, workspaceRepo, summaryRepo,
+        })
+
+        // Reuse repos for context_assembler and scheduler (replace Step 6 duplicate new)
+        this.context_assembler.set_data_sources({ message_repo: messageRepo, evidence_store: evidenceRepo })
+        this.scheduler.set_task_repo(taskRepo)
        const rehydrated = await this.scheduler.rebuild_from_db()
        this.logger.info('Scheduler recovery complete', { rehydrated })
      }
--- a/packages/runtime/src/scheduler/Scheduler.ts
+++ b/packages/runtime/src/scheduler/Scheduler.ts
@@ -62,7 +62,7 @@ export class Scheduler {
  /**
   * Create tasks from specifications.
   */
-  create_tasks(tasks: Array<{ id: TaskID; type: string; title: string; description?: string; depends_on?: string[] }>): void {
+  async create_tasks(tasks: Array<{ id: TaskID; type: string; title: string; description?: string; depends_on?: string[] }>): Promise<void> {
    for (const task of tasks) {
      this.graph.add_task({
        id: task.id,
@@ -71,9 +71,28 @@ export class Scheduler {
        description: task.description,
        dependencies: task.depends_on?.map(d => ({ task_id: d, type: 'hard' as const })) || []
      })
+
+      // Emit task.created events (INV-1: via event store for projection)
+      await eventIngestor.ingest({
+        id: `evt_${task.id}_created`,
+        type: 'task.created',
+        version: 1,
+        session_id: this.context.session_id,
+        project_id: this.context.project_id,
+        timestamp: new Date().toISOString(),
+        source: { kind: 'scheduler' },
+        route: ['scheduler', 'create'],
+        payload: {
+          task_id: task.id,
+          type: task.type,
+          title: task.title,
+          task_spec_json: { description: task.description || '' },
+          dependencies: (task.depends_on || []).map(d => ({ depends_on_task_id: d, dependency_type: 'hard', reason: '' })),
+          metadata: {},
+        }
+      })
    }

-    // Emit task.created events (INV-1: via projection, not direct status write)
    this.state = 'PLANNING_WAVE'
  }

@@ -175,6 +194,28 @@ export class Scheduler {
              })
              this.graph.update_status(task.id, 'running')
              this.agent_monitor.record_heartbeat(agent_id, task.id)
+
+              // INV-1: Emit agent.started event (durable) for projection
+              await eventIngestor.ingest({
+                id: `evt_${agent_id}_started`,
+                type: 'agent.started',
+                version: 1,
+                session_id: this.context.session_id,
+                project_id: this.context.project_id,
+                timestamp: new Date().toISOString(),
+                source: { kind: 'scheduler' },
+                route: ['scheduler', 'dispatch'],
+                payload: {
+                  agent_id,
+                  agent_type: task.type || 'executor',
+                  task_id: task.id,
+                  pid: 0,
+                  model_provider_id: '',
+                  model_id: '',
+                  workspace_id: `ws_${task.id}`,
+                  metadata: {},
+                }
+              })
            } catch {
              // INV-1: emit task.failed event for projection
              await eventIngestor.ingest({
@@ -278,6 +319,18 @@ export class Scheduler {
                }
              })
              this.graph.update_status(task.id, 'completed')
+              // INV-1: Emit agent.completed event (durable) for projection
+              await eventIngestor.ingest({
+                id: `evt_${handle.worker_id}_completed`,
+                type: 'agent.completed',
+                version: 1,
+                session_id: this.context.session_id,
+                project_id: this.context.project_id,
+                timestamp: new Date().toISOString(),
+                source: { kind: 'scheduler' },
+                route: ['scheduler', 'monitoring'],
+                payload: { agent_id: handle.worker_id, task_id: task.id, summary: result.summary, worker_result_ref: attempt_id, metadata: {} }
+              })
              this.agent_monitor.remove(handle.worker_id)
            } else if (result.status === 'blocked') {
              await eventIngestor.ingest({
@@ -320,6 +373,18 @@ export class Scheduler {
                payload: { task_id: task.id, agent_id: handle.worker_id, attempt_id, error: { message: result.summary }, evidence_refs: result.evidence_refs, metadata: { worker_status: result.status } }
              })
              this.graph.update_status(task.id, 'failed')
+              // INV-1: Emit agent.failed event (durable) for projection
+              await eventIngestor.ingest({
+                id: `evt_${handle.worker_id}_failed`,
+                type: 'agent.failed',
+                version: 1,
+                session_id: this.context.session_id,
+                project_id: this.context.project_id,
+                timestamp: new Date().toISOString(),
+                source: { kind: 'scheduler' },
+                route: ['scheduler', 'monitoring'],
+                payload: { agent_id: handle.worker_id, task_id: task.id, error: { message: result.summary }, evidence_refs: result.evidence_refs || [], metadata: {} }
+              })
              this.agent_monitor.remove(handle.worker_id)
            }
          }
--- a/packages/runtime/src/storage/repositories/AgentRepository.ts
+++ b/packages/runtime/src/storage/repositories/AgentRepository.ts
@@ -43,7 +43,7 @@ export type AgentInsert = Omit<AgentRecord, 'id' | 'status'> & {
  id?: AgentID
 }

-export type AgentUpdate = Partial<Omit<AgentRecord, 'id' | 'session_id' | 'started_at' | 'status'>>
+export type AgentUpdate = Partial<Omit<AgentRecord, 'id' | 'session_id' | 'started_at'>>

 // =============================================================================
 // AgentRepository
@@ -105,7 +105,11 @@ export class AgentRepository implements Repository<AgentRecord, AgentInsert, Age
    const fields: string[] = []
    const values: unknown[] = []

-    // NOTE: status is NOT updatable here - only EventStore.project() writes status columns
+    // status reaches here only via EventStore.project() (INV-1's authorized writer).
+    if (patch.status !== undefined) {
+      fields.push('status = ?')
+      values.push(patch.status)
+    }
    if (patch.pid !== undefined) {
      fields.push('pid = ?')
      values.push(patch.pid)
--- a/packages/runtime/src/storage/repositories/TaskAttemptRepository.ts
+++ b/packages/runtime/src/storage/repositories/TaskAttemptRepository.ts
@@ -44,7 +44,7 @@ export type TaskAttemptInsert = Omit<TaskAttemptRecord, 'id' | 'status'> & {
  id?: UUID
 }

-export type TaskAttemptUpdate = Partial<Omit<TaskAttemptRecord, 'id' | 'session_id' | 'task_id' | 'attempt_index' | 'started_at' | 'status'>>
+export type TaskAttemptUpdate = Partial<Omit<TaskAttemptRecord, 'id' | 'session_id' | 'task_id' | 'attempt_index' | 'started_at'>>

 // =============================================================================
 // TaskAttemptRepository
@@ -106,7 +106,11 @@ export class TaskAttemptRepository implements Repository<TaskAttemptRecord, Task
    const fields: string[] = []
    const values: unknown[] = []

-    // NOTE: status is NOT updatable here - only EventStore.project() writes status columns
+    // status reaches here only via EventStore.project() (INV-1's authorized writer).
+    if (patch.status !== undefined) {
+      fields.push('status = ?')
+      values.push(patch.status)
+    }
    if (patch.agent_id !== undefined) {
      fields.push('agent_id = ?')
      values.push(patch.agent_id)
--- a/packages/runtime/src/storage/repositories/TaskRepository.ts
+++ b/packages/runtime/src/storage/repositories/TaskRepository.ts
@@ -53,7 +53,10 @@ export type TaskInsert = Omit<TaskRecord, 'id' | 'status'> & {
  heartbeat_at?: ISOTimeString
 }

-export type TaskUpdate = Partial<Omit<TaskRecord, 'id' | 'session_id' | 'created_at' | 'status'>>
+// status IS updatable — but only reachable via EventStore.project() (INV-1).
+// project() is the sole caller of update(); guarding status here would block
+// the one authorized writer and freeze every row at its insert-time status.
+export type TaskUpdate = Partial<Omit<TaskRecord, 'id' | 'session_id' | 'created_at'>>

 // =============================================================================
 // TaskRepository
@@ -118,7 +121,11 @@ export class TaskRepository implements Repository<TaskRecord, TaskInsert, TaskUp
    const fields: string[] = []
    const values: unknown[] = []

-    // NOTE: status is NOT updatable here - only EventStore.project() writes status columns
+    // status reaches here only via EventStore.project() (INV-1's authorized writer).
+    if (patch.status !== undefined) {
+      fields.push('status = ?')
+      values.push(patch.status)
+    }
    if (patch.title !== undefined) {
      fields.push('title = ?')
      values.push(patch.title)
--- a/packages/runtime/src/storage/repositories/ToolRunRepository.ts
+++ b/packages/runtime/src/storage/repositories/ToolRunRepository.ts
@@ -48,7 +48,7 @@ export type ToolRunInsert = Omit<ToolRunRecord, 'id' | 'status'> & {
  id?: ToolRunID
 }

-export type ToolRunUpdate = Partial<Omit<ToolRunRecord, 'id' | 'session_id' | 'tool_name' | 'started_at' | 'status'>>
+export type ToolRunUpdate = Partial<Omit<ToolRunRecord, 'id' | 'session_id' | 'tool_name' | 'started_at'>>

 // =============================================================================
 // ToolRunRepository
@@ -114,7 +114,11 @@ export class ToolRunRepository implements Repository<ToolRunRecord, ToolRunInser
    const fields: string[] = []
    const values: unknown[] = []

-    // NOTE: status is NOT updatable here - only EventStore.project() writes status columns
+    // status reaches here only via EventStore.project() (INV-1's authorized writer).
+    if (patch.status !== undefined) {
+      fields.push('status = ?')
+      values.push(patch.status)
+    }
    if (patch.output_json !== undefined) {
      fields.push('output_json = ?')
      values.push(patch.output_json)
--- a/packages/runtime/src/tools/fs/index.ts
+++ b/packages/runtime/src/tools/fs/index.ts
@@ -8,9 +8,64 @@
 */

 import { readFileSync, writeFileSync, readdirSync, statSync, existsSync, mkdirSync } from 'fs'
+import { createHash } from 'crypto'
 import { join, dirname, basename, extname } from 'path'
 import type { ToolDefinition, ToolCall, ToolResultEnvelope, ISOTimeString } from '@aircoding/contracts'

+// =============================================================================
+// Read File State (for read-before-edit enforcement)
+// =============================================================================
+
+interface ReadFileState {
+  timestamp: number
+  sha256: string
+  size: number
+}
+
+// Session-scoped read file state - keyed by absolute path
+const read_file_state = new Map<string, ReadFileState>()
+
+function compute_sha256(content: string): string {
+  return createHash('sha256').update(content).digest('hex')
+}
+
+function record_file_read(abs_path: string, content: string): void {
+  read_file_state.set(abs_path, {
+    timestamp: Date.now(),
+    sha256: compute_sha256(content),
+    size: content.length
+  })
+}
+
+function check_file_read_state(abs_path: string, current_content: string): { allowed: boolean; error?: string } {
+  const state = read_file_state.get(abs_path)
+
+  if (!state) {
+    return {
+      allowed: false,
+      error: 'File has not been read yet. Read it first before editing.'
+    }
+  }
+
+  const current_sha = compute_sha256(current_content)
+  if (current_sha !== state.sha256) {
+    return {
+      allowed: false,
+      error: 'File has been unexpectedly modified. Read it again before editing.'
+    }
+  }
+
+  return { allowed: true }
+}
+
+function update_file_state(abs_path: string, new_content: string): void {
+  read_file_state.set(abs_path, {
+    timestamp: Date.now(),
+    sha256: compute_sha256(new_content),
+    size: new_content.length
+  })
+}
+
 // =============================================================================
 // Tool Definitions
 // =============================================================================
@@ -65,11 +120,13 @@ export const fs_edit: ToolDefinition = {
    type: 'object',
    properties: {
      path: { type: 'string', description: 'File path to edit' },
-      find: { type: 'string', description: 'Exact text to find' },
-      replace: { type: 'string', description: 'Text to replace with' },
-      global: { type: 'boolean', default: false, description: 'Replace all occurrences' }
+      find: { type: 'string', description: 'Exact text to find (alias: old_str)' },
+      replace: { type: 'string', description: 'Text to replace with (alias: new_str)' },
+      old_str: { type: 'string', description: 'Alias for find' },
+      new_str: { type: 'string', description: 'Alias for replace' },
+      global: { type: 'boolean', default: false, description: 'Replace all occurrences (alias: replace_all)' }
    },
-    required: ['path', 'find', 'replace']
+    required: ['path']
  },
  permissions: { read_paths: { allow: ["*"] }, write_paths: { allow: ["*"] } },
  streaming: false
@@ -153,6 +210,9 @@ export function createFsExecutors(project_root: string) {
          ? content.toString('base64')
          : content.toString('utf-8')

+        // Record file read for read-before-edit enforcement
+        record_file_read(full_path, content.toString('utf-8'))
+
        return create_result(call.call_id, 'fs.read', 'text', { content: output, size: content.length })
      } catch (error) {
        return create_result(call.call_id, 'fs.read', 'error', { message: error instanceof Error ? error.message : String(error) })
@@ -189,11 +249,15 @@ export function createFsExecutors(project_root: string) {
    },

    'fs.edit': async (call: ToolCall): Promise<ToolResultEnvelope> => {
-      const { path, find, replace, global = false } = call.arguments as {
-        path: string
-        find: string
-        replace: string
-        global?: boolean
+      // Support both old_str/new_str (ExecutorRole) and find/replace (UI) parameter names
+      const args = call.arguments as Record<string, unknown>
+      const path = args.path as string
+      const find = (args.find ?? args.old_str ?? '') as string
+      const replace = (args.replace ?? args.new_str ?? '') as string
+      const global = (args.global ?? args.replace_all ?? false) as boolean
+
+      if (!find) {
+        return create_result(call.call_id, 'fs.edit', 'error', { message: 'Missing find/old_str parameter' })
      }

      const full_path = resolve_path(path)
@@ -205,11 +269,25 @@ export function createFsExecutors(project_root: string) {
      try {
        const original = readFileSync(full_path, 'utf-8')

-        // Read-before-edit enforcement (DD §9.4)
+        // Read-before-edit enforcement (DD §9.4) - code layer, not prompt
+        const read_check = check_file_read_state(full_path, original)
+        if (!read_check.allowed) {
+          return create_result(call.call_id, 'fs.edit', 'error', { message: read_check.error })
+        }
+
+        // Exact edit: old_str must exist uniquely
        if (!original.includes(find)) {
          return create_result(call.call_id, 'fs.edit', 'error', { message: 'Exact text not found in file' })
        }

+        // Check for uniqueness when not global
+        if (!global) {
+          const matches = original.split(find)
+          if (matches.length > 2) {
+            return create_result(call.call_id, 'fs.edit', 'error', { message: 'Text appears multiple times. Use global=true or provide more context to make it unique.' })
+          }
+        }
+
        let edited: string
        if (global) {
          edited = original.split(find).join(replace)
@@ -219,6 +297,9 @@ export function createFsExecutors(project_root: string) {

        writeFileSync(full_path, edited, 'utf-8')

+        // Update read state after successful edit
+        update_file_state(full_path, edited)
+
        // Emit diff artifact (DD §9.4)
        return create_result(call.call_id, 'fs.edit', 'text', {
          message: `Edited ${path}`,