fix: SQLite UNIQUE constraint - event ID 冲突修复
问题:重复 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 <noreply@anthropic.com>
This commit is contained in:
@@ -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<string, unknown> }>): Promise<void> {
|
||||
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,
|
||||
|
||||
Reference in New Issue
Block a user