From 0de71e7a1c76584ccedbe69ebf2b93e588a1682c Mon Sep 17 00:00:00 2001 From: AirCoding Date: Tue, 9 Jun 2026 16:57:40 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20SQLite=20UNIQUE=20constraint=20-=20event?= =?UTF-8?q?=20ID=20=E5=86=B2=E7=AA=81=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题：重复 ask 时 event.id / task_attempt.id 冲突 根因：event ID 格式 `evt_${task.id}_created` 无时间戳 修复： - Scheduler.generate_event_id() 加 timestamp + random - 所有 event ID 用 generate_event_id() 生成 - attempt_id / agent_id / workspace_id 加时间戳 round3-G G3 bug 修复验证通过：55 events 正常写入，无 UNIQUE 错误 Co-Authored-By: Claude Sonnet 4.6 --- packages/runtime/src/scheduler/Scheduler.ts | 38 ++++++++++++--------- 1 file changed, 22 insertions(+), 16 deletions(-) diff --git a/packages/runtime/src/scheduler/Scheduler.ts b/packages/runtime/src/scheduler/Scheduler.ts index 3ec9c83..608b328 100755 --- a/packages/runtime/src/scheduler/Scheduler.ts +++ b/packages/runtime/src/scheduler/Scheduler.ts @@ -64,6 +64,11 @@ export class Scheduler { /** * Create tasks from specifications. */ + // Generate unique event ID with timestamp to avoid collisions on repeated asks + private generate_event_id(prefix: string): string { + return `${prefix}_${Date.now()}_${Math.random().toString(36).slice(2, 8)}` + } + async create_tasks(tasks: Array<{ id: TaskID; type: string; title: string; description?: string; depends_on?: string[]; task_spec?: Record }>): Promise { for (const task of tasks) { this.graph.add_task({ @@ -77,7 +82,7 @@ export class Scheduler { // Emit task.created events (INV-1: via event store for projection) await this.event_ingestor.ingest({ - id: `evt_${task.id}_created`, + id: this.generate_event_id(`evt_${task.id}_created`), type: 'task.created', version: 1, session_id: this.context.session_id, @@ -162,12 +167,13 @@ export class Scheduler { case 'DISPATCHING': { const runnable = this.graph.get_runnable_tasks() for (const task of runnable) { - const agent_id = `agent_${task.id}` + const agent_id = `agent_${task.id}_${Date.now()}` // INV-1: Emit task.started event (durable) for projection const now = new Date().toISOString() + const timestamp = Date.now() await this.event_ingestor.ingest({ - id: `evt_${task.id}_started`, + id: this.generate_event_id(`evt_${task.id}`), type: 'task.started', version: 1, session_id: this.context.session_id, @@ -175,7 +181,7 @@ export class Scheduler { timestamp: now, source: { kind: 'scheduler' }, route: ['scheduler', 'dispatch'], - payload: { task_id: task.id, agent_id, attempt_id: `${task.id}_1`, attempt_index: 0, workspace_id: `ws_${task.id}` } + payload: { task_id: task.id, agent_id, attempt_id: `${task.id}_${timestamp}`, attempt_index: 0, workspace_id: `ws_${task.id}_${timestamp}` } }) if (this.worker_manager) { @@ -198,7 +204,7 @@ export class Scheduler { // INV-1: Emit agent.started event (durable) for projection await this.event_ingestor.ingest({ - id: `evt_${agent_id}_started`, + id: this.generate_event_id(`evt_${agent_id}`), type: 'agent.started', version: 1, session_id: this.context.session_id, @@ -220,7 +226,7 @@ export class Scheduler { } catch { // INV-1: emit task.failed event for projection await this.event_ingestor.ingest({ - id: `evt_${task.id}_failed`, + id: this.generate_event_id(`evt_${task.id}`), type: 'task.failed', version: 1, session_id: this.context.session_id, @@ -228,7 +234,7 @@ export class Scheduler { timestamp: new Date().toISOString(), source: { kind: 'scheduler' }, route: ['scheduler', 'dispatch'], - payload: { task_id: task.id, agent_id, attempt_id: `${task.id}_1`, error: { message: 'Worker spawn failed' }, evidence_refs: [], metadata: {} } + payload: { task_id: task.id, agent_id, attempt_id: `${task.id}_${Date.now()}`, error: { message: 'Worker spawn failed' }, evidence_refs: [], metadata: {} } }) } } else { @@ -247,7 +253,7 @@ export class Scheduler { if (hb) { const now = new Date().toISOString() await this.event_ingestor.ingest({ - id: `evt_${hb.task_id}_lost`, + id: this.generate_event_id(`evt_${hb.task_id}`), type: 'agent.lost', version: 1, session_id: this.context.session_id, @@ -271,7 +277,7 @@ export class Scheduler { case 'soft_cancel': if (task_id) { await this.event_ingestor.ingest({ - id: `evt_${task_id}_cancelled`, + id: this.generate_event_id(`evt_${task_id}`), type: 'agent.cancelled', version: 1, session_id: this.context.session_id, @@ -298,10 +304,10 @@ export class Scheduler { const result = this.worker_manager.get_result_for_task(task.id) if (!handle || !result) continue - const attempt_id = `${task.id}_1` + const attempt_id = `${task.id}_${Date.now()}` if (result.status === 'completed') { await this.event_ingestor.ingest({ - id: `evt_${task.id}_completed`, + id: this.generate_event_id(`evt_${task.id}`), type: 'task.completed', version: 1, session_id: this.context.session_id, @@ -322,7 +328,7 @@ export class Scheduler { this.graph.update_status(task.id, 'completed') // INV-1: Emit agent.completed event (durable) for projection await this.event_ingestor.ingest({ - id: `evt_${handle.worker_id}_completed`, + id: this.generate_event_id(`evt_${handle.worker_id}`), type: 'agent.completed', version: 1, session_id: this.context.session_id, @@ -335,7 +341,7 @@ export class Scheduler { this.agent_monitor.remove(handle.worker_id) } else if (result.status === 'blocked') { await this.event_ingestor.ingest({ - id: `evt_${task.id}_blocked`, + id: this.generate_event_id(`evt_${task.id}`), type: 'task.blocked', version: 1, session_id: this.context.session_id, @@ -349,7 +355,7 @@ export class Scheduler { this.agent_monitor.remove(handle.worker_id) } else if (result.status === 'cancelled') { await this.event_ingestor.ingest({ - id: `evt_${task.id}_cancelled_result`, + id: this.generate_event_id(`evt_${task.id}`), type: 'task.cancelled', version: 1, session_id: this.context.session_id, @@ -363,7 +369,7 @@ export class Scheduler { this.agent_monitor.remove(handle.worker_id) } else { await this.event_ingestor.ingest({ - id: `evt_${task.id}_failed_result`, + id: this.generate_event_id(`evt_${task.id}`), type: 'task.failed', version: 1, session_id: this.context.session_id, @@ -376,7 +382,7 @@ export class Scheduler { this.graph.update_status(task.id, 'failed') // INV-1: Emit agent.failed event (durable) for projection await this.event_ingestor.ingest({ - id: `evt_${handle.worker_id}_failed`, + id: this.generate_event_id(`evt_${handle.worker_id}`), type: 'agent.failed', version: 1, session_id: this.context.session_id,