Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 25 additions & 5 deletions loopx/control_plane/coordination/todo_create.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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]) =>
Expand All @@ -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},
);
}
Expand Down Expand Up @@ -165,15 +166,34 @@ export function planCoordinationTodoCreate(
rawInput: CoordinationTodoCreateInput,
todos: ReadonlyMap<string, JsonObject>,
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});
}
Expand Down
3 changes: 2 additions & 1 deletion loopx/control_plane/scheduler/monitor_batch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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});
Expand Down
65 changes: 65 additions & 0 deletions tests/control_plane_ts/todo_monitor_poll.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
Loading