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
23 changes: 23 additions & 0 deletions docs/reference/todo-work-counts.md
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,29 @@ evaluation still precedes filtering, and returned positions refer to the origina
array, not a newly numbered subset. Python retains public field allowlists,
warning text, privacy redaction and Markdown parsing/rendering.

Agent summaries compute their existing `advancement_frontier_revision_index`
inside this same projection, after source selection and before display caps.
The shared frontier owner retains the v0 serialization/hash, excluded-agent
lanes, owned-work ACK identity and commitment counts. Missing or contradictory
codec facts fail the read; no separate index RPC repairs the result. The
single-caller Python attachment is retired. Direct frontier calls used by replan
and successor causality remain, as do Markdown writers for supported unmigrated
Goals, backup/receipt recovery and Host IO. Internal summary request v3 composes
the existing owners; v1/v2 callers and public summary/index schemas remain
compatible. This is a source-read retirement slice, not SQLite-default or full
writer-retirement qualification. Roll back the matching Python/TS distribution
together; no persisted data migration is involved.

Agent summary 在同一 TS 批次中生成既有 `advancement_frontier_revision_index`:先筛选
完整来源,再生成索引,最后应用展示上限。沿用 frontier owner 的 v0 序列化/哈希、
排除 Agent 的 lane、已拥有工作对应的 ACK 身份和承诺计数。缺失或矛盾的 codec 事实会
明确拒绝读取,不会补调另一个索引 RPC。删除仅有一个调用方的 Python attachment;
replan 和 successor 因果核验仍使用的直接 frontier 入口、未迁移 Goal 的 Markdown
writer、备份/原回执恢复及 Host IO 继续保留。内部 summary 请求升级为 v3,保留
v1/v2 调用兼容和公共 summary/index schema。本批只完成来源读取边界的退役,不代表
SQLite 默认或全部 writer 退役已验收;回退时一起回退匹配的 Python/TS 分发,无需迁移
持久化数据。

**Intentional read behavior changes:** recent completions are ordered by the
actual `completed_at` instant, preserving timezone offsets and microseconds.
Later `updated_at` edits no longer make an old completion recent. Missing or
Expand Down
7 changes: 0 additions & 7 deletions loopx/control_plane/todos/frontier_revision.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,10 +103,3 @@ def build_advancement_frontier_revision_index(
if not isinstance(index, dict):
raise TypeError("typed frontier revision response index must be an object")
return index


def attach_advancement_frontier_revision_index(
summary: dict[str, Any], source_items: list[dict[str, Any]], *, role: str | None,
) -> None:
if role == "agent":
summary["advancement_frontier_revision_index"] = build_advancement_frontier_revision_index(source_items)
31 changes: 29 additions & 2 deletions loopx/control_plane/todos/frontier_revision.ts
Original file line number Diff line number Diff line change
Expand Up @@ -232,15 +232,42 @@ export function projectAdvancementFrontier(value: unknown): JsonObject {
const rows = decodeRows(request.rows);
if (request.operation === "select") return {checkpoint: checkpoint(rows, agent)};
if (request.operation !== "index") throw new EffectRuntimeRequestError("unsupported frontier revision operation");
return {index: frontierIndex(rows)};
}

function frontierIndex(rows: Row[] | null): JsonObject {
// An excluded agent can have no claimed rows. It still needs its own lane;
// falling back to the global unclaimed checkpoint would include excluded work.
const agents = [...new Set((rows ?? []).filter(row => row.advancement)
.flatMap(row => [...(row.claim ? [row.claim] : []), ...row.excluded]))].sort();
const counts = claimedAdvancementCounts(rows);
return {index: {schema_version: INDEX, ...(counts === null ? {} : {claimed_advancement_counts: counts}),
return {schema_version: INDEX, ...(counts === null ? {} : {claimed_advancement_counts: counts}),
all: checkpoint(rows, null),
unclaimed: checkpoint(rows, null, true),
by_agent: agents.map(agent_id => ({agent_id, ...checkpoint(rows, agent_id)}))}};
by_agent: agents.map(agent_id => ({agent_id, ...checkpoint(rows, agent_id)}))};
}

/** Compose the existing index owner inside an already-required summary read.
* Bind the lossless legacy codec to the validated source before selecting;
* display limits must never decide replan identity or commitment counts. */
export function projectSummaryFrontierIndex(value: unknown, source: readonly JsonObject[],
indices: readonly number[]): JsonObject {
const rows = decodeRows(value);
if (rows === null || rows.length !== source.length) {
throw new EffectRuntimeRequestError("summary frontier source cardinality mismatch");
}
for (const [index, row] of rows.entries()) {
const fact = source[index];
const actionable = fact.status === "open" && !fact.done &&
(!fact.has_resume || fact.resume_ready === true) && !fact.acceptance_blocked;
if (row.id !== text(fact.todo_id) || row.claim !== (text(fact.claim) || null) ||
JSON.stringify(row.excluded) !== JSON.stringify(fact.excluded) ||
row.advancement !== (fact.task_class === "advancement_task") ||
row.actionable !== actionable || row.updated !== text(fact.updated_at || fact.completed_at)) {
throw new EffectRuntimeRequestError("summary frontier facts disagree with the validated source");
}
}
return frontierIndex(indices.map(index => rows[index]));
}

function classifyAck(observation: LongChainObservation, value: unknown): AckDecision {
Expand Down
11 changes: 10 additions & 1 deletion loopx/control_plane/todos/summary_projection.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
import {requireBoolean, requireJsonObject, requireStringLiteral} from "../runtime_decode.ts";
import {parseTodoTimestampMicros} from "../runtime_timestamp.ts";
import {projectTodoSummaryLanes, type TodoSummaryLane} from "./summary_lanes.ts";
import {projectSummaryFrontierIndex} from "./frontier_revision.ts";
import {projectTodoClosure, projectTodoSuccession, SUCCESSION_FACT_COLUMNS, SUCCESSION_EVALUATION_COLUMNS} from "./succession.ts";

type Format = "raw" | "compact" | "active" | "recent" | "gap";
Expand Down Expand Up @@ -100,7 +101,8 @@ function claimedVisibility(indices: readonly number[], rows: readonly JsonObject
export function projectTodoSummary(value: unknown): SummaryProjection {
const request = requireJsonObject(value, "Todo summary request");
if (request.schema_version !== "todo_summary_projection_request_v1" &&
request.schema_version !== "todo_summary_projection_request_v2") {
request.schema_version !== "todo_summary_projection_request_v2" &&
request.schema_version !== "todo_summary_projection_request_v3") {
throw new EffectRuntimeRequestError("Todo summary request schema mismatch");
}
const role = request.role === null ? null : requireStringLiteral(request.role, ["user", "agent"], "role");
Expand Down Expand Up @@ -142,6 +144,13 @@ export function projectTodoSummary(value: unknown): SummaryProjection {
monitor_schedule_gap_count: selected.monitor_schedule_gap_items.length,
};
const lanes: Record<string, DisplayLane> = {};
if (request.schema_version === "todo_summary_projection_request_v3") {
if (role === "agent") {
fields.advancement_frontier_revision_index = projectSummaryFrontierIndex(request.frontier_rows, rows, source);
} else if (request.frontier_rows !== null) {
throw new EffectRuntimeRequestError("non-Agent summary must omit frontier source facts");
}
}
const lane = (name: string, indices: readonly number[], cap: number | null = null, format: Format = "compact") => {
lanes[name] = {indices: cap === null ? [...indices] : indices.slice(0, cap), format};
};
Expand Down
10 changes: 7 additions & 3 deletions loopx/control_plane/todos/todo_summary.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@
todo_done_for_status,
)
from .completion_validation_projection import project_completion_validation_authority
from .frontier_revision import attach_advancement_frontier_revision_index
from .frontier_revision import frontier_source_facts, TODO_FRONTIER_REVISION_INDEX_SCHEMA_VERSION
from .handoff_gate import build_todo_handoff_gate_states
from .handoff_note import attach_todo_handoff_note
from .todo_semantics import (
Expand Down Expand Up @@ -84,7 +84,7 @@
# facts: repeating every key name per Todo pushed a long-history request past the
# effect-runtime request budget. The typed owner decodes the declared columns
# back into row objects before validating them, so no cell changes meaning.
SUMMARY_PROJECTION_REQUEST_SCHEMA_VERSION = "todo_summary_projection_request_v2"
SUMMARY_PROJECTION_REQUEST_SCHEMA_VERSION = "todo_summary_projection_request_v3"
SUMMARY_PROJECTION_COLUMNS = (
"status", "done", "task_class", "has_resume", "resume_ready", "resume_evaluated",
"acceptance_blocked", "claimed", "preferred", "watch_only", "due_at", "expires_at",
Expand Down Expand Up @@ -874,6 +874,7 @@ def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] |
"columns": list(SUMMARY_PROJECTION_COLUMNS),
"rows": [[row[name] for name in SUMMARY_PROJECTION_COLUMNS] for row in rows],
"succession": succession_request(items, reuse=True),
"frontier_rows": frontier_source_facts(items) if role == "agent" else None,
"observed_at": now_utc().timestamp(),
"selection": selection, "role": role, "source_section": source_section,
"item_limit": item_limit, "full_selection": full_selection,
Expand Down Expand Up @@ -906,6 +907,10 @@ def valid_ordinals(value: Any) -> TypeGuard[list[int]]:
summary = result.get("fields")
if not isinstance(summary, dict) or summary.get("schema_version") != "todo_summary_v0":
raise ValueError("invalid typed Todo summary fields")
if role == "agent":
index = summary.get("advancement_frontier_revision_index")
if not isinstance(index, dict) or index.get("schema_version") != TODO_FRONTIER_REVISION_INDEX_SCHEMA_VERSION:
raise ValueError("invalid typed Todo summary frontier index")
warning = summary.get("todo_succession_warning")
warning_action = warning.get("recommended_action") if isinstance(warning, dict) else None
if summary.get("completed_without_successor_count") and not isinstance(warning_action, str):
Expand Down Expand Up @@ -1012,7 +1017,6 @@ def compact_evaluated_todo_group(
return None
summary: dict[str, Any] = projected["summary"]
handoff_gates = build_todo_handoff_gate_states(items, evaluations=projected["succession"])
attach_advancement_frontier_revision_index(summary, items, role=role)
attach_active_vision_waits(
summary, vision_runs, role=role, items=items,
lineage_items=lineage_items,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ def cli(command, *args, expected_exit=0):
assert not state.exists(), "rejected recovery cannot revive a legacy writer"
readback = cli(["todo", "list"])
assert first["source_authority"] == provider + "_v0"
assert readback["todos"][0]["claimed_by"] == ("agent-b" if retirement == "transfer" else "agent-a")
assert readback["todo"]["claimed_by"] == ("agent-b" if retirement == "transfer" else "agent-a")
assert not (runtime / "goals" / goal / "task-leases" / f"{target}.json").exists()
finally:
subprocess.run([sys.executable, "-c", "from loopx.control_plane.effect_runtime import effect_runtime_result; effect_runtime_result('runtime.shutdown',{},retry_safe=False)"],
Expand Down
6 changes: 3 additions & 3 deletions tests/control_plane/test_canonical_claim_transfer.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ def cli(command, *args, expected_exit=0):
assert result["source_authority"] == provider + "_v0"
assert result["projection_delivery"] == "pending"
assert result["projection_outbox"]["retry_business_mutation"] is False
assert cli(["todo", "list"])["todos"][0]["claimed_by"] == "agent-b"
assert cli(["todo", "list"])["todo"]["claimed_by"] == "agent-b"
state.rmdir()
state.write_text(original_text)
replay = cli(["task-lease", "transfer"], *transfer, "--transfer-claim")
Expand All @@ -82,13 +82,13 @@ def cli(command, *args, expected_exit=0):
"--task-lease-expected-version", "2", "--evidence", "validation://claim-transfer", "--no-follow-up")
assert complete["ok"]
current = cli(["todo", "list"])
assert current["todos"][0]["status"] == "done"
assert current["todo"]["status"] == "done"
current_lease = cli(["task-lease", "inspect"])
assert current_lease["lease"]["status"] == "released" and not current_lease["active"]
historical = cli(["task-lease", "transfer"], *transfer, "--transfer-claim")
assert historical["original_receipt"] == result["original_receipt"]
assert historical["lease"]["status"] == "active"
assert cli(["todo", "list"])["todos"][0]["status"] == "done"
assert cli(["todo", "list"])["todo"]["status"] == "done"
assert cli(["task-lease", "inspect"])["lease"] == current_lease["lease"]
assert not (runtime / "goals" / goal / "task-leases" / f"{target}.json").exists()
finally:
Expand Down
2 changes: 1 addition & 1 deletion tests/control_plane/test_canonical_lease_acquire.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ def cli(command, *args, expected_exit=0):
assert completed["ok"]
assert state.exists()
readback = cli(["todo", "list"])
assert readback["todos"][0]["status"] == "done"
assert readback["todo"]["status"] == "done"
terminal = cli(["task-lease", "inspect"])
assert terminal["lease"]["status"] == "released" and not terminal["active"]
assert (cli(["task-lease", "acquire"], *next_proof, "--ttl-seconds", "600", expected_exit=1))["error_code"] == "todo_not_open"
Expand Down
5 changes: 3 additions & 2 deletions tests/control_plane/test_canonical_todo_writer_isolation.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,8 +72,9 @@ def add(role, text, operation):
"--note", "Retain source identity", "--update-operation-id", "update",
"--update-expected-provider-revision", listing["authority_read"]["provider_revision"])
assert updated["todo_id"] == original
updated_records = cli("list", "--goal-id", "goal-a", "--todo-id", original)["todos"]
assert updated_records[0]["note"] == "Retain source identity"
updated_record = cli("list", "--goal-id", "goal-a", "--todo-id", original)["todo"]
assert updated_record["todo_id"] == original
assert updated_record["note"] == "Retain source identity"
completed = cli("complete", "--goal-id", "goal-a", "--todo-id", user,
"--no-follow-up", "--note", "Independent user action accepted")
assert completed["completed"] and completed["todo_id"] == user
Expand Down
4 changes: 2 additions & 2 deletions tests/control_plane/test_causal_blocked_closeout_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]:
assert rc == 0 and guard["should_run"] is True, guard
rc, original = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID)
assert rc == 0, original
digest = original["todos"][0]["completion_validation_sha256"]
digest = original["todo"]["completion_validation_sha256"]
rc, wait = cli("todo", "update", "--goal-id", GOAL_ID,
"--todo-id", TODO_ID, "--agent-id", AGENT_ID,
"--resume-when", f"{kind}:{MONITOR_ID}",
Expand Down Expand Up @@ -121,7 +121,7 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]:
assert _spend_run_count(runtime) == 0
rc, after = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID)
assert rc == 0, after
todo = after["todos"][0]
todo = after["todo"]
assert todo["status"] == ("deferred" if defer else "open") and todo["resume_ready"] is False
assert todo["resume_when"] == f"{kind}:{MONITOR_ID}"
assert todo["completion_validation_sha256"] == digest
Expand Down
4 changes: 2 additions & 2 deletions tests/control_plane/test_long_history_todo_reads.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ def test_public_list_preserves_completed_dependency_after_archive(tmp_path: Path
code, result = run_json_cli_result("todo", "list", "--goal-id", "goal-a", "--role", "agent",
"--todo-id", "todo_successor", "--agent-id", "agent-a", registry_path=registry, runtime_root=runtime)
assert code == 0, result
assert result["todos"][0]["resume_ready"] is True
assert result["todo"]["resume_ready"] is True
assert state.read_bytes() == after


Expand Down Expand Up @@ -85,7 +85,7 @@ def test_capture_preserves_archived_dependency_for_real_canonical_cli_read(tmp_p
code, result = run_json_cli_result("todo", "list", "--goal-id", "goal-a", "--role", "agent",
"--todo-id", "todo_waiting", registry_path=registry, runtime_root=runtime)
assert code == 0, result
assert result["todos"][0]["resume_ready"] is True
assert result["todo"]["resume_ready"] is True
code, completed = run_json_cli_result("todo", "list", "--goal-id", "goal-a", "--role", "agent",
"--status", "done", registry_path=registry, runtime_root=runtime)
assert code == 0, completed
Expand Down
9 changes: 6 additions & 3 deletions tests/control_plane/test_shadow_registry_source.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,11 +149,14 @@ def test_terminal_lease_survives_public_cutover_and_original_recovery(tmp_path:
original = json.dumps(expected, indent=2).encode()
lease.write_bytes(original)
if provider == "sqlite":
subprocess.run([os.environ.get("LOOPX_CONTROL_PLANE_NODE", "node"),
selection = subprocess.run([os.environ.get("LOOPX_CONTROL_PLANE_NODE", "node"),
"--no-warnings", "--experimental-strip-types", "--experimental-sqlite",
"loopx/control_plane/coordination/local_authority_provider.ts", "--runtime-root",
str((REPO / "loopx/control_plane/coordination/local_authority_provider.ts").resolve()), "--runtime-root",
str(ws.runtime), "--goal-id", ws.goal, "--execute"], cwd=REPO,
capture_output=True, text=True, timeout=45, check=True)
assert json.loads(selection.stdout) == {
"ok": True, "provider": "sqlite", "changed": True, "executed": True,
}
assert ws.cli("coordination-shadow", "bootstrap", "--execute")["bootstrap"]["status"] == "applied"
for index in range(3):
ws.add(f"Independent migration work {index}")
Expand All @@ -171,7 +174,7 @@ def test_terminal_lease_survives_public_cutover_and_original_recovery(tmp_path:
assert recovered["promotion"]["status"] == "replayed", recovered
assert ws.cli("task-lease", "inspect", "--todo-id", todo_id)["lease"] == expected
retained = ws.cli("todo", "list", "--todo-id", later["todo_id"])
assert retained["todos"][0]["text"] == "Later canonical work must survive recovery"
assert retained["todo"]["text"] == "Later canonical work must survive recovery"
assert lease.read_bytes() == original
assert not ws.state.exists()

Expand Down
Loading
Loading