diff --git a/loopx/control_plane/coordination/todo_create.ts b/loopx/control_plane/coordination/todo_create.ts index e5c30401b6..45a62d7fc6 100644 --- a/loopx/control_plane/coordination/todo_create.ts +++ b/loopx/control_plane/coordination/todo_create.ts @@ -109,6 +109,7 @@ function semanticDuplicateResult( duplicate: JsonObject, providerRevision: unknown, cursor: unknown, + identityDescription = "role/text", ): CoordinationTodoCreateResult { const ignored = new Set(["schema_version", "todo_id", "created_by", "last_actor_agent_id", "updated_at"]); const mismatch = Object.entries(todo).find(([field, value]) => @@ -123,7 +124,7 @@ function semanticDuplicateResult( if (mismatch !== undefined) { return failure( "todo_semantic_duplicate_conflict", - `an active Todo with the same role/text has different ${mismatch[0]}; use Todo update`, + `an active Todo with the same ${identityDescription} has different ${mismatch[0]}; use Todo update`, {todo_id: duplicate.todo_id}, ); } @@ -165,15 +166,34 @@ export function planCoordinationTodoCreate( rawInput: CoordinationTodoCreateInput, todos: ReadonlyMap, readModelSchema: unknown, - identity: "role_text" | "operation_lane" = "role_text", + identity: "role_text" | "role_target" | "operation_lane" = "role_text", claimPolicy: "actor_owned" | "registered_peer_handoff" = "actor_owned", ): CoordinationTodoCreateResult { const input = normalizeCreateInput(rawInput, claimPolicy); const candidate = createCandidate(input, readModelSchema); - const duplicate = identity === "role_text" ? [...todos.values()].find((todo) => + const active = [...todos.values()].filter((todo) => todo.role === input.todo.role && todo.archive_state === "active" && - todo.status !== "done" && todo.status !== "deferred" && todo.text === candidate.text) : undefined; - if (identity === "role_text" && duplicate !== undefined) return semanticDuplicateResult(candidate, duplicate, null, null); + todo.status !== "done" && todo.status !== "deferred"); + const textDuplicate = identity === "role_text" + ? active.find((todo) => todo.text === candidate.text) + : undefined; + if (textDuplicate !== undefined) { + return semanticDuplicateResult(candidate, textDuplicate, null, null); + } + if (identity === "role_target" && + (typeof candidate.target_key !== "string" || candidate.target_key.length === 0)) { + return failure("todo_target_identity_missing", "target-key Todo identity requires target_key"); + } + const targetDuplicates = identity === "role_target" + ? active.filter((todo) => todo.target_key === candidate.target_key) + : []; + if (targetDuplicates.length > 1) { + return failure("todo_semantic_duplicate_ambiguous", + "multiple active Todos have the same role/target_key"); + } + if (targetDuplicates.length === 1) { + return semanticDuplicateResult(candidate, targetDuplicates[0]!, null, null, "role/target_key"); + } if (todos.has(String(input.todo.todo_id))) { return failure("todo_already_exists", "Todo id already exists in canonical authority", {todo_id: input.todo.todo_id}); } diff --git a/loopx/control_plane/scheduler/monitor_batch.ts b/loopx/control_plane/scheduler/monitor_batch.ts index af4434d4c8..c57d242adb 100644 --- a/loopx/control_plane/scheduler/monitor_batch.ts +++ b/loopx/control_plane/scheduler/monitor_batch.ts @@ -122,7 +122,8 @@ function planWriteback(input: MonitorBatchInput) { const created = planCoordinationTodoCreate({goal_id: input.goal_id, operation_id: input.operation_id, actor_agent_id: actor, registered_agents: input.registered_agents, dry_run: input.dry_run, now: new Date(String(observation.generated_at)), todo}, plannedTodos, - input.read_model_schema ?? TODO_DOMAIN_READ_RECORD_SCHEMA, "role_text", "actor_owned"); + input.read_model_schema ?? TODO_DOMAIN_READ_RECORD_SCHEMA, + role === "agent" && intent.next_target_key ? "role_target" : "role_text", "actor_owned"); if (created.status !== "planned" && created.status !== "no_change") throw new EffectRuntimeRequestError(String(created.reason)); const record = canonicalAuthorityObject(created.todo, "successor Todo"); nextTodos.push({...record, todo: record.text, ok: true, dry_run: input.dry_run}); diff --git a/tests/control_plane_ts/todo_monitor_poll.test.ts b/tests/control_plane_ts/todo_monitor_poll.test.ts index c3519202d9..5b1f128ffe 100644 --- a/tests/control_plane_ts/todo_monitor_poll.test.ts +++ b/tests/control_plane_ts/todo_monitor_poll.test.ts @@ -54,6 +54,71 @@ test("monitor and independent successor commit together, preview is inert, recei "coordination_operation_identity_mismatch"); }); +test("explicit successor target reuses one compatible open Todo across material generations", async () => { + const {store, request} = await seeded(); + const intent = {...request.intent, next_target_key: "deliver-watch-a"}; + const first = await executeCoordinationMonitorPoll(store, {...request, intent}); + assert.equal(first.status, "applied", JSON.stringify(first)); + const firstWriteback = first.writeback as JsonObject; + const firstTodo = (firstWriteback.next_todos as JsonObject[])[0]!; + const secondRequest = {...request, operation_id: "poll-b", intent, + observation: {...request.observation, generated_at: "2026-09-01T01:00:00Z", result_hash: "result-b"}}; + const second = await executeCoordinationMonitorPoll(store, secondRequest); + assert.equal(second.status, "applied", JSON.stringify(second)); + const secondWriteback = second.writeback as JsonObject; + assert.equal(secondWriteback.material_change_generation, 2); + assert.equal((secondWriteback.next_todos as JsonObject[])[0]!.todo_id, firstTodo.todo_id); + assert.equal((secondWriteback.successor_receipts as JsonObject[])[0]!.todo_id, firstTodo.todo_id); + assert.deepEqual( + (secondWriteback.next_todos as JsonObject[]).map(todo => todo.todo_id), + (secondWriteback.successor_receipts as JsonObject[]).map(todo => todo.todo_id), + ); + const after = await store.loadAuthority(); + assert.equal(after.status, "loaded"); + if (after.status !== "loaded") return; + assert.equal((after.head.todos as JsonObject[]).length, 2); + assert.equal((after.head.todos as JsonObject[]).filter(todo => todo.target_key === "deliver-watch-a").length, 1); + assert.equal((await executeCoordinationMonitorPoll(store, secondRequest)).status, "replayed"); + assert.deepEqual(await store.loadAuthority(), after); +}); + +test("explicit successor target rejects incompatible text or route without updating the Monitor", async () => { + const {store, request} = await seeded(); + const intent = {...request.intent, next_target_key: "deliver-watch-a"}; + assert.equal((await executeCoordinationMonitorPoll(store, {...request, intent})).status, "applied"); + const before = await store.loadAuthority(); + for (const [operation_id, result_hash, intentChange, mismatch] of [ + ["poll-text-conflict", "result-b", {next_agent_todo: "Deliver a different result"}, "text"], + ["poll-route-conflict", "result-c", {next_action_kind: "validate"}, "action_kind"], + ] as const) { + const result = await executeCoordinationMonitorPoll(store, {...request, operation_id, + observation: {...request.observation, generated_at: "2026-09-01T01:00:00Z", result_hash}, + intent: {...intent, ...intentChange}}); + assert.equal(result.status, "failed"); + assert.equal(result.reason_code, "monitor_poll_rejected"); + assert.match(String(result.reason), new RegExp(`role/target_key.*${mismatch}`)); + assert.deepEqual(await store.loadAuthority(), before); + assert.equal((await store.readReceipt(operation_id)).status, "missing"); + } +}); + +test("explicit successor target rejects ambiguous active identity without updating the Monitor", async () => { + const duplicate = (todo_id: string): JsonObject => ({schema_version: TODO_DOMAIN_ITEM_SCHEMA, todo_id, + role: "agent", status: "open", done: false, text: "Deliver the newly observed result", + archive_state: "active", task_class: "advancement_task", action_kind: "implementation", + continuation_policy: "independent_handoff", target_key: "deliver-watch-a", + unblocks_todo_id: "todo_monitor"}); + const {store, request} = await seeded({}, [duplicate("todo_z_existing_a"), duplicate("todo_z_existing_b")]); + const before = await store.loadAuthority(); + const result = await executeCoordinationMonitorPoll(store, {...request, + intent: {...request.intent, next_target_key: "deliver-watch-a"}}); + assert.equal(result.status, "failed"); + assert.equal(result.reason_code, "monitor_poll_rejected"); + assert.match(String(result.reason), /multiple active Todos have the same role\/target_key/); + assert.deepEqual(await store.loadAuthority(), before); + assert.equal((await store.readReceipt(request.operation_id)).status, "missing"); +}); + test("receipt-only recovery preserves the original hash and never plans fresh work", async () => { const {store, request} = await seeded(); const before = await store.loadAuthority();