From b53edec56d47d8f8de087f9487c7552511b7d199 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 20:45:05 +0800 Subject: [PATCH 1/2] fix(todo): publish event-owned completion and successors atomically Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../control_plane/effect_runtime_handlers.ts | 2 + .../control_plane/goals/state_event_append.ts | 68 +++++ loopx/control_plane/todos/event_writeback.py | 264 +++++++----------- loopx/event_sourced_state.py | 157 ++++++++--- .../test_event_todo_transaction.py | 189 +++++++++++++ .../test_todo_mutation_authority.py | 10 +- .../state_event_append.test.ts | 59 ++++ tests/test_event_sourced_state_store.py | 213 +++++++++++++- 8 files changed, 760 insertions(+), 202 deletions(-) create mode 100644 loopx/control_plane/goals/state_event_append.ts create mode 100644 tests/control_plane/test_event_todo_transaction.py create mode 100644 tests/control_plane_ts/state_event_append.test.ts diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 8c86154f08..14477e78a3 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -196,6 +196,7 @@ import {evaluateCapabilityGate} from "./agents/capability_gate.ts"; import {projectCoordinationSource} from "./coordination/source_projection.ts"; import {withCoordinationSourceTransfer} from "./coordination/source_transfer.ts"; import {captureArchivedTodoDependencies} from "./todos/archive_capture.ts"; +import {planStateEventAppend} from "./goals/state_event_append.ts"; import {projectAdvancementFrontier, evaluateLongTodoChain} from "./todos/frontier_revision.ts"; import { evaluateCoordinationTodoSuccessorDerivation } from "./coordination/todo_successor_derivation.ts"; import { @@ -455,6 +456,7 @@ export function createEffectRuntimeHandlers( ["agent.capability_gate.evaluate", evaluateCapabilityGate], ["agent.capability_memory", agentCapabilityMemory], ["todo.archive.capture_dependencies", withCoordinationSourceTransfer("todo.archive.capture_dependencies", captureArchivedTodoDependencies)], + ["goal.state_event.plan_append", planStateEventAppend], ["coordination.source.project", withCoordinationSourceTransfer("coordination.source.project", projectCoordinationSource)], ["todo.monitor_metadata.plan", planMonitorMetadata], ["todo.authoring_scope.plan", planTodoAuthoringScope], diff --git a/loopx/control_plane/goals/state_event_append.ts b/loopx/control_plane/goals/state_event_append.ts new file mode 100644 index 0000000000..9e5bbfbabb --- /dev/null +++ b/loopx/control_plane/goals/state_event_append.ts @@ -0,0 +1,68 @@ +/** Legacy event-log append admission. Python owns the locked bytes and codec; + * this owner allocates sequences and rejects a whole conflicting/stale batch. */ +import type {JsonObject} from "../effect_program.ts"; +import {requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; +import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; + +type EventIdentity = {event_id: string; fingerprint: string}; +type StoredIdentity = EventIdentity & {append_sequence: number}; +type AppendChoice = {kind: "replay" | "append"; event_id: string; append_sequence: number}; + +function sequence(value: unknown, minimum: number): number { + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < minimum) { + throw new EffectRuntimeRequestError("event append sequence must be a safe integer"); + } + return value; +} + +function identity(value: unknown): EventIdentity { + const row = requireJsonObject(value, "event identity"); + const fingerprint = requireNonEmptyString(row.fingerprint, "event fingerprint"); + if (!/^[a-f0-9]{64}$/.test(fingerprint)) throw new EffectRuntimeRequestError("invalid event fingerprint"); + return {event_id: requireNonEmptyString(row.event_id, "event_id"), fingerprint}; +} + +export function planStateEventAppend(value: unknown): JsonObject { + const request = requireJsonObject(value, "state event append plan"); + if (request.schema_version !== "loopx_state_event_append_plan_v0" || + !Array.isArray(request.existing) || !Array.isArray(request.events)) { + throw new EffectRuntimeRequestError("invalid state event append plan"); + } + requireNonEmptyString(request.source_checksum, "source_checksum"); + if (request.expected_checksum !== null) { + requireNonEmptyString(request.expected_checksum, "expected_checksum"); + } + const result = (status: "planned" | "rejected", fields: JsonObject): JsonObject => + ({schema_version: "loopx_state_event_append_result_v0", status, ...fields}); + if (request.expected_checksum !== null && request.expected_checksum !== request.source_checksum) { + return result("rejected", {reason_code: "event_source_changed"}); + } + let last = sequence(request.last_sequence, 0); + const known = new Map(); + for (const raw of request.existing) { + const row = requireJsonObject(raw, "stored event identity"); + const record = {...identity(row), append_sequence: sequence(row.append_sequence, 1)}; + if (record.append_sequence > last || known.has(record.event_id)) { + throw new EffectRuntimeRequestError("inconsistent stored event identity"); + } + known.set(record.event_id, record); + } + const choices: AppendChoice[] = []; + for (const raw of request.events) { + const item = identity(raw), prior = known.get(item.event_id); + if (prior) { + if (prior.fingerprint !== item.fingerprint) { + return result("rejected", {reason_code: "event_id_conflict", event_id: item.event_id}); + } + choices.push({kind: "replay", event_id: item.event_id, append_sequence: prior.append_sequence}); + } else { + if (last === Number.MAX_SAFE_INTEGER) { + return result("rejected", {reason_code: "event_sequence_exhausted"}); + } + const record = {...item, append_sequence: ++last}; + known.set(item.event_id, record); + choices.push({kind: "append", event_id: item.event_id, append_sequence: record.append_sequence}); + } + } + return result("planned", {choices}); +} diff --git a/loopx/control_plane/todos/event_writeback.py b/loopx/control_plane/todos/event_writeback.py index db24a15ac9..de69d37f2d 100644 --- a/loopx/control_plane/todos/event_writeback.py +++ b/loopx/control_plane/todos/event_writeback.py @@ -12,6 +12,7 @@ from ...event_sourced_state import ( AppendOnlyStateEventStore, StateEventError, + StateEventSourceChangedError, TODO_ADDED, TODO_CLAIMED, TODO_COMPLETED, @@ -27,19 +28,12 @@ from ..runtime.validation_command import CALLER_VALIDATION_RECEIPT_SCHEMA_VERSION from .active_state_todo_parser import parse_active_state_todos from .contract import ( - TODO_CONTINUATION_POLICY_VALUES, TODO_STATUS_DONE, build_todo_id, merge_todo_id_lists, - normalize_required_capabilities, - normalize_todo_capability_binding_ref, normalize_todo_claimed_by, - normalize_todo_bound_agent, - normalize_todo_continuation_policy, - normalize_todo_excluded_agents, normalize_todo_id, normalize_todo_id_list, - normalize_todo_task_repository, ) from .completion_transaction import reduce_todo_completion_transaction from .successor_derivation import ( @@ -47,10 +41,6 @@ derive_successor_proposals, ) from .todo_semantics import todo_priority_parts -from .text import ( - normalize_new_todo, - todo_priority_prefix, -) TODO_SECTION_HEADINGS = { @@ -277,165 +267,101 @@ def _todo_write_event_id( return f"todo-write-{action}-{digest}" -def _append_event_projected_successor( +def _encode_event_projected_successor( *, - store: AppendOnlyStateEventStore, goal_id: str, - role: str, - text: str, + proposal: Mapping[str, Any], updated_at: str, fields: dict[str, Any], - task_class: str | None, - action_kind: str | None, - capability_binding_ref: str | None, - task_repository: str | None, - required_capabilities: list[str] | None, - continuation_policy: str | None, - claimed_by: str | None, - dry_run: bool, - actor_agent_id: str | None = None, - bound_agent: str | None = None, - goal_bound: bool | None = None, - blocks_agent: str | None = None, - excluded_agents: list[str] | None = None, - unblocks_todo_id: str | None = None, -) -> dict[str, Any]: + actor_agent_id: str | None, +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + """Encode a TS-derived successor without deciding defaults or writing it.""" + role = str(proposal["role"]) section = TODO_SECTION_HEADINGS[role] summary = fields.get(f"{role}_todos") items = summary.get("items") if isinstance(summary, dict) else [] index = len(items if isinstance(items, list) else []) + 1 - todo_text = normalize_new_todo(text) - todo_id = build_todo_id( - role=role, - source_section=section, - index=index, - text=todo_text, + text = str(proposal["text"]) + todo_id = build_todo_id(role=role, source_section=section, index=index, text=text) + priority, title = todo_priority_parts(text) + metadata_fields = ( + "task_class", + "action_kind", + "capability_binding_ref", + "task_repository", + "required_capabilities", + "continuation_policy", + "bound_agent", + "goal_bound", + "blocks_agent", + "excluded_agents", + "unblocks_todo_id", ) - _priority, title = todo_priority_parts(todo_text) - payload: dict[str, Any] = { + metadata = { + key: proposal[key] for key in metadata_fields + if key in proposal and proposal[key] not in (None, "", []) + } + payload = { "role": role, - "priority": todo_priority_prefix(todo_text) or "P2", + "priority": priority or "P2", "title": title, "planner_order": index, "updated_at": updated_at, + **metadata, } - if task_class: - payload["task_class"] = task_class - if action_kind: - payload["action_kind"] = action_kind - normalized_capability_binding_ref = normalize_todo_capability_binding_ref( - capability_binding_ref - ) - if capability_binding_ref and not normalized_capability_binding_ref: - raise ValueError( - "capability_binding_ref must be a public-safe namespaced token" - ) - if normalized_capability_binding_ref: - payload["capability_binding_ref"] = normalized_capability_binding_ref - normalized_task_repository = normalize_todo_task_repository(task_repository) - if task_repository and not normalized_task_repository: - raise ValueError( - "todo task_repository must be a credential-free Git remote or canonical " - "git:/ identity" - ) - if normalized_task_repository: - payload["task_repository"] = normalized_task_repository - normalized_required_capabilities = normalize_required_capabilities( - required_capabilities - ) - if required_capabilities and not normalized_required_capabilities: - raise ValueError( - "required_capabilities must contain public-safe capability tokens" - ) - if normalized_required_capabilities: - payload["required_capabilities"] = normalized_required_capabilities - normalized_continuation_policy = normalize_todo_continuation_policy( - continuation_policy - ) - if continuation_policy and not normalized_continuation_policy: - raise ValueError( - "todo continuation_policy must be one of: " - + ", ".join(sorted(TODO_CONTINUATION_POLICY_VALUES)) - ) - effective_continuation_policy = normalized_continuation_policy - if effective_continuation_policy: - payload["continuation_policy"] = effective_continuation_policy - if blocks_agent: - payload["blocks_agent"] = blocks_agent - if bound_agent: - payload["bound_agent"] = normalize_todo_bound_agent(bound_agent) - if goal_bound is not None: - payload["goal_bound"] = goal_bound - normalized_excluded_agents = normalize_todo_excluded_agents(excluded_agents) - if claimed_by in normalized_excluded_agents: - raise ValueError( - f"claimed_by={claimed_by!r} cannot also appear in excluded_agents" - ) - if normalized_excluded_agents: - payload["excluded_agents"] = normalized_excluded_agents - if unblocks_todo_id: - payload["unblocks_todo_id"] = unblocks_todo_id - added_event = make_state_event( - event_id=_todo_write_event_id( - goal_id=goal_id, - todo_id=todo_id, - action="add", - updated_at=updated_at, - text=todo_text, - ), - goal_id=goal_id, - event_type=TODO_ADDED, - refs={"todo_id": todo_id}, - payload=payload, - recorded_at=updated_at, - producer="loopx.todo.complete", - actor_agent_id=actor_agent_id, - ) - claimed_event: dict[str, Any] | None = None - if claimed_by: - claimed_event = make_state_event( + events = [ + make_state_event( event_id=_todo_write_event_id( goal_id=goal_id, todo_id=todo_id, - action="claim", + action="add", updated_at=updated_at, - text=claimed_by, + text=text, ), goal_id=goal_id, - event_type=TODO_CLAIMED, + event_type=TODO_ADDED, refs={"todo_id": todo_id}, - payload={"claimed_by": claimed_by}, + payload=payload, recorded_at=updated_at, producer="loopx.todo.complete", actor_agent_id=actor_agent_id, ) - if not dry_run: - store.append(added_event) - if claimed_event: - store.append(claimed_event) + ] + claimed_by = proposal.get("claimed_by") + if claimed_by: + events.append( + make_state_event( + event_id=_todo_write_event_id( + goal_id=goal_id, + todo_id=todo_id, + action="claim", + updated_at=updated_at, + text=claimed_by, + ), + goal_id=goal_id, + event_type=TODO_CLAIMED, + refs={"todo_id": todo_id}, + payload={"claimed_by": claimed_by}, + recorded_at=updated_at, + producer="loopx.todo.complete", + actor_agent_id=actor_agent_id, + ) + ) return { "added": True, "already_exists": False, "metadata_updated": False, "role": role, "section": section, - "todo": todo_text, + "todo": text, "todo_id": todo_id, - "task_class": task_class, - "action_kind": action_kind, - "capability_binding_ref": normalized_capability_binding_ref, - "task_repository": normalized_task_repository, - "required_capabilities": normalized_required_capabilities, - "continuation_policy": effective_continuation_policy, + **{key: proposal.get(key) for key in metadata_fields}, + "required_capabilities": proposal.get("required_capabilities", []), + "excluded_agents": proposal.get("excluded_agents", []), "claimed_by": claimed_by, - "bound_agent": bound_agent, - "goal_bound": goal_bound, - "blocks_agent": blocks_agent, - "excluded_agents": normalized_excluded_agents, - "unblocks_todo_id": unblocks_todo_id, "updated_at": updated_at, "source": "event_log", - } + }, events def complete_event_projected_goal_todo( @@ -474,14 +400,25 @@ def complete_event_projected_goal_todo( registry_path = Path(context["registry_path"]) state_path = Path(context["state_path"]) root = runtime_root or effective_runtime_root(registry_path, None) - transaction = nullcontext() if primary_lock_held else legacy_todo_write_transaction( - registry_path, goal_id, state_path, actor_agent_id, "todo_event_complete", - dry_run, runtime_root=root, + transaction = ( + nullcontext() + if primary_lock_held + else legacy_todo_write_transaction( + registry_path, + goal_id, + state_path, + actor_agent_id, + "todo_event_complete", + dry_run, + runtime_root=root, + ) ) with transaction: if not dry_run: require_shadow_primary_write_allowed(root, goal_id) - require_legacy_coordination_write_allowed(runtime_root=root, goal_id=goal_id) + require_legacy_coordination_write_allowed( + runtime_root=root, goal_id=goal_id + ) item = dict(context["item"]) role = str(context["role"]) todo_id = normalize_todo_id(item.get("todo_id")) @@ -498,8 +435,11 @@ def complete_event_projected_goal_todo( ) if clear_claim and item.get("claimed_by"): item.pop("claimed_by", None) - effective_claimed_by = claimed_by or normalize_todo_claimed_by(item.get("claimed_by")) + effective_claimed_by = claimed_by or normalize_todo_claimed_by( + item.get("claimed_by") + ) store = AppendOnlyStateEventStore(Path(context["event_log_path"])) + source_checksum = context["fields"]["state_event_projection"]["source_checksum"] if completion_fence is None or completion_state is None: transaction = reduce_todo_completion_transaction( todo=item, @@ -525,9 +465,7 @@ def complete_event_projected_goal_todo( completion_fence = dict(transaction["fence"]) candidate_state = transaction.get("completion_state") completion_state = ( - dict(candidate_state) - if isinstance(candidate_state, Mapping) - else None + dict(candidate_state) if isinstance(candidate_state, Mapping) else None ) already_done = bool(completion_fence["terminal_before_request"]) terminal_upgrade = completion_fence["reason"] in { @@ -541,6 +479,13 @@ def complete_event_projected_goal_todo( completion_fence["reason"] == "unscoped_completion_identity_repair" ) if completion_fence["outcome"] == "replay": + if not dry_run: + try: + store.append_many([], expected_checksum=source_checksum) + except StateEventSourceChangedError: + return _completion_validation_source_drift_failure( + goal_id=goal_id, todo_id=todo_id, dry_run=dry_run + ) return { "ok": True, "dry_run": dry_run, @@ -588,30 +533,18 @@ def complete_event_projected_goal_todo( }, successor_intents=successor_intents, ) - next_results = [ - _append_event_projected_successor( - store=store, + encoded_successors = [ + _encode_event_projected_successor( goal_id=goal_id, - role=str(proposal["role"]), - text=str(proposal["text"]), + proposal=proposal, updated_at=updated_at, fields=context["fields"], - task_class=proposal.get("task_class"), - action_kind=proposal.get("action_kind"), - capability_binding_ref=proposal.get("capability_binding_ref"), - task_repository=proposal.get("task_repository"), - required_capabilities=proposal.get("required_capabilities"), - continuation_policy=proposal.get("continuation_policy"), - claimed_by=proposal.get("claimed_by"), - bound_agent=proposal.get("bound_agent"), - blocks_agent=proposal.get("blocks_agent"), - excluded_agents=proposal.get("excluded_agents"), - unblocks_todo_id=proposal.get("unblocks_todo_id"), - dry_run=dry_run, actor_agent_id=actor_agent_id, ) for proposal in successor_proposals ] + next_results = [result for result, _ in encoded_successors] + pending_events = [event for _, events in encoded_successors for event in events] normalized_successor_todo_ids = merge_todo_id_lists( successor_todo_ids, @@ -674,8 +607,15 @@ def complete_event_projected_goal_todo( or terminal_upgrade or untyped_completion_repair or unscoped_identity_repair - ) and not dry_run: - store.append(completion_event) + ): + pending_events.append(completion_event) + if not dry_run: + try: + store.append_many(pending_events, expected_checksum=source_checksum) + except StateEventSourceChangedError: + return _completion_validation_source_drift_failure( + goal_id=goal_id, todo_id=todo_id, dry_run=dry_run + ) result = { "ok": True, diff --git a/loopx/event_sourced_state.py b/loopx/event_sourced_state.py index e05d9a6d54..e9a84b76bb 100644 --- a/loopx/event_sourced_state.py +++ b/loopx/event_sourced_state.py @@ -108,6 +108,14 @@ class StateEventConflictError(StateEventError): """Raised when a duplicate event id carries different event content.""" +class StateEventSourceChangedError(StateEventError): + """The locked event stream differs from the basis used to plan the write.""" + + +class StateEventCommitUnknownError(StateEventError): + """Publication may have landed; read back before repeating business work.""" + + def now_utc_iso() -> str: return runtime_now_utc_iso() @@ -583,48 +591,127 @@ def load(self) -> list[dict[str, Any]]: def append(self, event: dict[str, Any]) -> dict[str, Any]: return self.append_many((event,))[0] - def append_many(self, events: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: + def append_many( + self, + events: Iterable[dict[str, Any]], + *, + expected_checksum: str | None = None, + ) -> list[dict[str, Any]]: + """Publish an eager batch atomically; lazy iterables retain per-item visibility. + + A compare-and-append requires an eager batch. Materialize caller-owned + iterators outside this method when one atomic publication is intended. + """ if type(events) not in (list, tuple): + if expected_checksum is not None: + raise StateEventError("a source-bound append requires a list or tuple") return [self.append(event) for event in events] - if not events: + if not events and expected_checksum is None: return [] + from .control_plane.effect_runtime import effect_runtime_result + from .control_plane.todos.active_state_editing import ( + atomic_write_state_text, + verify_state_text_durable, + ) + + # Validate all caller data before entering the lock or publishing any + # bytes. Sequence allocation is deliberately deferred to the TS owner. + normalized = [ + normalize_state_event(event, append_sequence=1) for event in events + ] + requested_ids = {item["event_id"] for item in normalized} + + def identity(item: dict[str, Any]) -> dict[str, Any]: + return { + "event_id": item["event_id"], + "fingerprint": hashlib.sha256( + event_fingerprint(item).encode("utf-8") + ).hexdigest(), + } + with exclusive_file_lock(self.path): stored = self.load() existing = {item["event_id"]: item for item in stored} - next_sequence = max( - (int(item["append_sequence"]) for item in stored), default=0 - ) + 1 - appended: list[dict[str, Any]] = [] - stream = None - try: - for event in events: - normalized = normalize_state_event( - event, - append_sequence=next_sequence, + plan = effect_runtime_result( + "goal.state_event.plan_append", + { + "schema_version": "loopx_state_event_append_plan_v0", + "source_checksum": event_stream_checksum( + sorted(stored, key=event_sort_key) + ), + "expected_checksum": expected_checksum, + "last_sequence": max( + (int(item["append_sequence"]) for item in stored), default=0 + ), + "existing": [ + {**identity(item), "append_sequence": item["append_sequence"]} + for item in stored + if item["event_id"] in requested_ids + ], + "events": [identity(item) for item in normalized], + }, + ) + if plan.get("schema_version") != "loopx_state_event_append_result_v0": + raise StateEventError("invalid event append plan result schema") + if plan.get("status") != "planned": + reason = plan.get("reason_code") + if reason == "event_source_changed": + raise StateEventSourceChangedError( + "event source changed before append" ) - prior = existing.get(normalized["event_id"]) - if prior is not None: - if event_fingerprint(prior) != event_fingerprint(normalized): - raise StateEventConflictError( - f"conflicting event_id: {normalized['event_id']}" - ) - appended.append(prior) - continue - - if stream is None: - self.path.parent.mkdir(parents=True, exist_ok=True) - stream = self.path.open("a", encoding="utf-8") - stream.write( - json.dumps(normalized, sort_keys=True, ensure_ascii=False) + "\n" + if reason == "event_id_conflict": + raise StateEventConflictError( + f"conflicting event_id: {plan['event_id']}" ) - stream.flush() - existing[normalized["event_id"]] = normalized - appended.append(normalized) - next_sequence += 1 - finally: - if stream is not None: - stream.close() + raise StateEventError(str(reason or "invalid event append plan")) + choices = plan.get("choices") + if not isinstance(choices, list) or len(choices) != len(normalized): + raise StateEventError("invalid event append plan choices") + appended: list[dict[str, Any]] = [] + additions: list[dict[str, Any]] = [] + for event, choice in zip(normalized, choices, strict=True): + if ( + not isinstance(choice, dict) + or choice.get("event_id") != event["event_id"] + or choice.get("kind") not in {"append", "replay"} + or isinstance(choice.get("append_sequence"), bool) + or not isinstance(choice.get("append_sequence"), int) + or not 1 <= choice["append_sequence"] <= 2**53 - 1 + ): + raise StateEventError("invalid event append plan choice") + if choice["kind"] == "append": + event["append_sequence"] = choice["append_sequence"] + existing[event["event_id"]] = event + additions.append(event) + appended.append(existing[event["event_id"]]) + # Preserve all historical bytes, including harmless blank lines. + # Replacing the whole file changes no prior event or sequence. + prior_text = ( + self.path.read_bytes().decode("utf-8") if self.path.exists() else "" + ) + if additions: + separator = "\n" if prior_text and not prior_text.endswith("\n") else "" + suffix = "".join( + json.dumps(item, sort_keys=True, ensure_ascii=False) + "\n" + for item in additions + ) + try: + atomic_write_state_text(self.path, prior_text + separator + suffix) + except OSError as error: + raise StateEventCommitUnknownError( + "event append outcome uncertain; read back the event stream before retrying the original operation" + ) from error + else: + # A prior replace may have succeeded before directory fsync + # failed. Exact replay must establish durability, not just see it. + if self.path.exists(): + try: + verify_state_text_durable(self.path, prior_text) + except OSError as error: + raise StateEventCommitUnknownError( + "event replay durability uncertain; read back the event stream before retrying" + ) from error return appended @@ -635,7 +722,9 @@ def _dedupe_events(events: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: prior = by_id.get(event["event_id"]) if prior is not None: if event_fingerprint(prior) != event_fingerprint(event): - raise StateEventConflictError(f"conflicting event_id: {event['event_id']}") + raise StateEventConflictError( + f"conflicting event_id: {event['event_id']}" + ) continue by_id[event["event_id"]] = event ordered.append(event) diff --git a/tests/control_plane/test_event_todo_transaction.py b/tests/control_plane/test_event_todo_transaction.py new file mode 100644 index 0000000000..8296d05798 --- /dev/null +++ b/tests/control_plane/test_event_todo_transaction.py @@ -0,0 +1,189 @@ +"""The public completion entrypoint must never publish half a continuation.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from loopx.control_plane.todos import active_state_editing as io +from loopx.control_plane.todos import event_writeback +from loopx.event_sourced_state import ( + TODO_ADDED, + TODO_COMPLETED, + TODO_UPDATED, + AppendOnlyStateEventStore, + StateEventCommitUnknownError, + build_state_projection, + make_state_event, +) +from loopx.todos import complete_goal_todo + + +@pytest.fixture +def event_goal(tmp_path: Path): + state = tmp_path / "ACTIVE_GOAL_STATE.md" + state.write_text( + "---\ngoal_id: event-transaction\n---\n\n## Agent Todo\n", encoding="utf-8" + ) + registry = tmp_path / "registry.json" + registry.write_text( + json.dumps( + { + "common_runtime_root": str(tmp_path / "runtime"), + "goals": [ + { + "id": "event-transaction", + "status": "active", + "repo": str(tmp_path), + "state_file": state.name, + "domain": "harness_self_improvement", + "adapter": {"kind": "harness_self_improvement"}, + "coordination": { + "agent_model": "peer_v1", + "registered_agents": ["worker"], + }, + } + ], + } + ), + encoding="utf-8", + ) + store = AppendOnlyStateEventStore(tmp_path / "events.jsonl") + store.append( + make_state_event( + event_id="parent", + goal_id="event-transaction", + event_type=TODO_ADDED, + refs={"todo_id": "todo_parent"}, + payload={ + "role": "agent", + "title": "Validate the implementation before independent review.", + "task_class": "advancement_task", + "claimed_by": "worker", + }, + recorded_at="2026-09-24T00:00:00Z", + ) + ) + request = dict( + registry_path=registry, + goal_id="event-transaction", + todo_id="todo_parent", + claimed_by="worker", + evidence="Implementation validation passed.", + completion_turn_key="completion-transaction", + next_agent_todo="Review the implementation independently.", + next_task_class="advancement_task", + next_claimed_by="worker", + ) + return store, request, state + + +def test_late_completion_encoding_failure_cannot_orphan_successor( + event_goal, monkeypatch +): + store, request, state = event_goal + before, markdown = store.path.read_bytes(), state.read_bytes() + encode = event_writeback.make_state_event + + def fail_completion(**kwargs): + if kwargs["event_type"] == TODO_COMPLETED: + raise ValueError("injected completion encoding failure") + return encode(**kwargs) + + monkeypatch.setattr(event_writeback, "make_state_event", fail_completion) + with pytest.raises(ValueError, match="encoding failure"): + complete_goal_todo(**request) + assert store.path.read_bytes() == before + assert state.read_bytes() == markdown + + +def test_concurrent_source_change_rejects_entire_completion(event_goal, monkeypatch): + store, request, _ = event_goal + derive = event_writeback.derive_successor_proposals + + def concurrent_update(**kwargs): + proposals = derive(**kwargs) + store.append( + make_state_event( + event_id="concurrent-update", + goal_id="event-transaction", + event_type=TODO_UPDATED, + refs={"todo_id": "todo_parent"}, + payload={"title": "A revised validation requirement."}, + recorded_at="2026-09-24T00:01:00Z", + ) + ) + return proposals + + monkeypatch.setattr( + event_writeback, "derive_successor_proposals", concurrent_update + ) + result = complete_goal_todo(**request) + assert result["ok"] is False + assert result["completed"] is False + assert [row["event_id"] for row in store.load()] == ["parent", "concurrent-update"] + + +def test_lost_sync_ack_retries_public_completion_without_duplicate_work( + event_goal, monkeypatch +): + store, request, _ = event_goal + sync = io.fsync_state_directory + failures = [] + + def fail_event_log_once(path): + if path == store.path and not failures: + failures.append(path) + raise OSError("injected event-log directory sync failure") + return sync(path) + + monkeypatch.setattr(io, "fsync_state_directory", fail_event_log_once) + with pytest.raises(StateEventCommitUnknownError): + complete_goal_todo(**request) + landed = store.path.read_bytes() + projection = build_state_projection(store.load()) + todos = projection["agent_todos"]["items"] + parent = next(row for row in todos if row["todo_id"] == "todo_parent") + assert parent["status"] == "done" + assert len(parent["successor_todo_ids"]) == 1 + assert parent["successor_todo_ids"][0] in {row["todo_id"] for row in todos} + replay = complete_goal_todo(**request) + assert replay["idempotent_replay"] is True + assert replay["changed"] is False + assert store.path.read_bytes() == landed + + +def test_complete_successors_are_visible_together_and_dry_run_is_read_only(event_goal): + store, request, state = event_goal + request.update( + next_user_todo="Approve the proposed delivery.", + next_user_task_class="user_gate", + ) + before, markdown = store.path.read_bytes(), state.read_bytes() + preview = complete_goal_todo(**request, dry_run=True) + assert preview["ok"] is True + assert store.path.read_bytes() == before + result = complete_goal_todo(**request) + assert result["ok"] is True + assert len(result["next_todos"]) == 2 + for successor in result["next_todos"]: + assert successor["required_capabilities"] == [] + assert successor["excluded_agents"] == [] + events = store.load() + assert events[-1]["event_type"] == TODO_COMPLETED + projection = build_state_projection(events) + parent = next( + row + for row in projection["agent_todos"]["items"] + if row["todo_id"] == "todo_parent" + ) + all_ids = { + row["todo_id"] + for role in ("agent_todos", "user_todos") + for row in projection[role]["items"] + } + assert len(parent["successor_todo_ids"]) == 2 + assert set(parent["successor_todo_ids"]) <= all_ids + assert state.read_bytes() == markdown diff --git a/tests/control_plane/test_todo_mutation_authority.py b/tests/control_plane/test_todo_mutation_authority.py index c683835e94..da91d47c08 100644 --- a/tests/control_plane/test_todo_mutation_authority.py +++ b/tests/control_plane/test_todo_mutation_authority.py @@ -1762,7 +1762,10 @@ def test_capability_binding_follows_event_projected_successor(tmp_path: Path) -> "item": parent, "role": "agent", "event_log_path": event_log, - "fields": {"agent_todos": projection["agent_todos"]}, + "fields": { + "agent_todos": projection["agent_todos"], + "state_event_projection": {"source_checksum": projection["source_checksum"]}, + }, }, evidence="bounded validation passed", note=None, @@ -1813,7 +1816,10 @@ def test_capability_binding_follows_event_projected_successor(tmp_path: Path) -> "item": completed_parent, "role": "agent", "event_log_path": event_log, - "fields": {"agent_todos": replayed["agent_todos"]}, + "fields": { + "agent_todos": replayed["agent_todos"], + "state_event_projection": {"source_checksum": replayed["source_checksum"]}, + }, }, evidence="late stale completion", note=None, diff --git a/tests/control_plane_ts/state_event_append.test.ts b/tests/control_plane_ts/state_event_append.test.ts new file mode 100644 index 0000000000..6cd909c1b4 --- /dev/null +++ b/tests/control_plane_ts/state_event_append.test.ts @@ -0,0 +1,59 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {planStateEventAppend} from "../../loopx/control_plane/goals/state_event_append.ts"; + +const old = {event_id: "old", fingerprint: "a".repeat(64), append_sequence: 7}; +const fresh = {event_id: "new", fingerprint: "b".repeat(64)}; +const request = {schema_version: "loopx_state_event_append_plan_v0", source_checksum: "basis", + expected_checksum: "basis", last_sequence: 7, existing: [old], events: [fresh, old, fresh]}; + +test("allocate only once for a duplicate, preserving historical replay order", () => { + assert.deepEqual(planStateEventAppend(request).choices, [ + {kind: "append", event_id: "new", append_sequence: 8}, + {kind: "replay", event_id: "old", append_sequence: 7}, + {kind: "replay", event_id: "new", append_sequence: 8}, + ]); + assert.equal(request.last_sequence, 7); +}); + +test("late conflicts never return a partial append plan", () => { + for (const event_id of ["old", "new"]) { + const result = planStateEventAppend({...request, + events: [fresh, {event_id, fingerprint: "c".repeat(64)}]}); + assert.equal(result.status, "rejected"); + assert.equal(result.reason_code, "event_id_conflict"); + assert.equal(result.choices, undefined); + } +}); + +test("source mismatch wins even for an empty durability confirmation", () => { + const result = planStateEventAppend({...request, source_checksum: "changed", events: []}); + assert.equal(result.reason_code, "event_source_changed"); + assert.equal(result.choices, undefined); +}); + +test("sequence exhaustion cannot round two new events onto one identity", () => { + assert.equal(planStateEventAppend({...request, last_sequence: Number.MAX_SAFE_INTEGER}).reason_code, + "event_sequence_exhausted"); + assert.throws(() => planStateEventAppend({...request, last_sequence: Number.MAX_SAFE_INTEGER + 1})); + assert.equal(planStateEventAppend({...request, last_sequence: Number.MAX_SAFE_INTEGER, events: [old]}).status, + "planned"); +}); + +test("malformed or contradictory compact source witnesses reject", () => { + for (const changed of [{source_checksum: null}, {expected_checksum: undefined}, {last_sequence: true}, {existing: [old, old]}, + {existing: [{...old, append_sequence: 8}]}, {events: [{...fresh, fingerprint: "not-a-hash"}]}]) { + assert.throws(() => planStateEventAppend({...request, ...changed})); + } +}); + +test("valid event ids cannot collide with JavaScript object prototype names", () => { + const events = ["__proto__", "constructor", "toString"].map(event_id => ({...fresh, event_id})); + assert.deepEqual(planStateEventAppend({...request, existing: [], last_sequence: 0, + events: [...events, events[0]]}).choices, [ + {kind: "append", event_id: "__proto__", append_sequence: 1}, + {kind: "append", event_id: "constructor", append_sequence: 2}, + {kind: "append", event_id: "toString", append_sequence: 3}, + {kind: "replay", event_id: "__proto__", append_sequence: 1}, + ]); +}); diff --git a/tests/test_event_sourced_state_store.py b/tests/test_event_sourced_state_store.py index 062b516f53..e09bd8c903 100644 --- a/tests/test_event_sourced_state_store.py +++ b/tests/test_event_sourced_state_store.py @@ -171,7 +171,7 @@ def __iter__(self): ] -def test_append_many_flushes_each_event_before_processing_the_next( +def test_eager_batch_has_no_visible_prefix_while_validating_later_events( tmp_path: Path, ) -> None: event_log = tmp_path / "events.jsonl" @@ -189,7 +189,7 @@ def test_append_many_flushes_each_event_before_processing_the_next( class ObservingEvent(dict): def get(self, key, default=None): if not observed_prefix: - observed_prefix.append(event_log.read_text(encoding="utf-8")) + observed_prefix.append(event_log.read_text(encoding="utf-8") if event_log.exists() else "") return super().get(key, default) second = ObservingEvent( @@ -205,11 +205,101 @@ def get(self, key, default=None): store.append_many([first, second]) - assert '"event_id": "flush-event-0"' in observed_prefix[0] + assert observed_prefix == [""] + assert len(store.load()) == 2 + + +def _event(identity: str, title: str = "A complete event") -> dict: + return make_state_event( + event_id=identity, + goal_id="goal-a", + event_type=TODO_ADDED, + refs={"todo_id": f"todo_{identity}"}, + payload={"role": "agent", "title": title}, + recorded_at="2026-09-24T00:00:00Z", + ) + + +@pytest.mark.parametrize("failure", ["invalid", "stored_conflict", "batch_conflict"]) +def test_late_batch_failure_preserves_every_prior_byte( + tmp_path: Path, failure: str +) -> None: + store = AppendOnlyStateEventStore(tmp_path / "events.jsonl") + store.append(_event("existing")) + before = store.path.read_bytes() + tail = ( + {"event_id": "invalid"} + if failure == "invalid" + else _event( + "existing" if failure == "stored_conflict" else "new", "Conflicting payload" + ) + ) + with pytest.raises(StateEventError): + store.append_many([_event("new"), tail]) + assert store.path.read_bytes() == before + + +def test_stale_basis_cannot_publish_successor_prefix(tmp_path: Path) -> None: + store = AppendOnlyStateEventStore(tmp_path / "events.jsonl") + store.append(_event("original")) + checksum = event_sourced_state.event_stream_checksum(store.load()) + store.append(_event("concurrent")) + before = store.path.read_bytes() + with pytest.raises(event_sourced_state.StateEventSourceChangedError): + store.append_many([_event("successor")], expected_checksum=checksum) + assert store.path.read_bytes() == before + + +def test_lost_publication_ack_replays_without_new_events_and_syncs_again( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from loopx.control_plane.todos import active_state_editing as io + + store = AppendOnlyStateEventStore(tmp_path / "events.jsonl") + sync = io.fsync_state_directory + attempts = [] + + def fail_once(path): + attempts.append(path) + if len(attempts) == 1: + raise OSError("injected directory sync failure after replacement") + return sync(path) + + monkeypatch.setattr(io, "fsync_state_directory", fail_once) + batch = [_event("first"), _event("second")] + with pytest.raises(event_sourced_state.StateEventCommitUnknownError): + store.append_many(batch) + landed = store.path.read_bytes() + assert [item["event_id"] for item in store.load()] == ["first", "second"] + assert [item["append_sequence"] for item in store.append_many(batch)] == [1, 2] + assert store.path.read_bytes() == landed + assert len(attempts) == 2 + + +def test_replay_durability_failure_remains_an_uncertain_commit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + from loopx.control_plane.todos import active_state_editing as io + + store = AppendOnlyStateEventStore(tmp_path / "events.jsonl") + event = _event("first") + store.append(event) + before = store.path.read_bytes() + + def fail_sync(path: Path) -> None: + raise OSError("injected replay sync failure") + + monkeypatch.setattr(io, "fsync_state_directory", fail_sync) + with pytest.raises(event_sourced_state.StateEventCommitUnknownError): + store.append(event) + assert store.path.read_bytes() == before @pytest.mark.parametrize("sequence", [True, False, 1.5, "2"]) -def test_load_rejects_non_integer_append_sequence(tmp_path: Path, sequence: object) -> None: +def test_load_rejects_non_integer_append_sequence( + tmp_path: Path, sequence: object +) -> None: event_log = tmp_path / "events.jsonl" event = make_state_event( event_id="evt-bool-sequence", @@ -224,3 +314,118 @@ def test_load_rejects_non_integer_append_sequence(tmp_path: Path, sequence: obje with pytest.raises(StateEventError, match="append_sequence must be an integer"): AppendOnlyStateEventStore(event_log).load() + + +@pytest.mark.parametrize("ending", [b"\r\n\r\n", b""]) +def test_atomic_append_preserves_historical_bytes( + tmp_path: Path, ending: bytes +) -> None: + store = AppendOnlyStateEventStore(tmp_path / "events.jsonl") + first = store.append(_event("first")) + historical = json.dumps(first, ensure_ascii=False).encode() + ending + store.path.write_bytes(historical) + store.append(_event("second")) + assert store.path.read_bytes().startswith(historical) + assert len(store.load()) == 2 + store.append(_event("first")) + assert len(store.load()) == 2 + + +def test_failure_before_replace_leaves_old_stream_intact( + tmp_path: Path, monkeypatch +) -> None: + from loopx.control_plane.todos import active_state_editing as io + + store = AppendOnlyStateEventStore(tmp_path / "events.jsonl") + store.append(_event("first")) + before = store.path.read_bytes() + + def fail_replace(*args): + raise OSError("injected pre-publication failure") + + monkeypatch.setattr(io.os, "replace", fail_replace) + with pytest.raises(event_sourced_state.StateEventCommitUnknownError): + store.append_many([_event("second"), _event("third")]) + assert store.path.read_bytes() == before + assert not list(tmp_path.glob(".*.tmp")) + + +@pytest.mark.parametrize( + "choice", + [ + {"kind": "unknown", "event_id": "second", "append_sequence": 2}, + {"kind": "append", "event_id": "wrong", "append_sequence": 2}, + {"kind": "append", "event_id": "second", "append_sequence": True}, + ], +) +def test_invalid_native_append_reply_does_not_publish( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + choice: dict, +) -> None: + from loopx.control_plane import effect_runtime + + store = AppendOnlyStateEventStore(tmp_path / "events.jsonl") + store.append(_event("first")) + before = store.path.read_bytes() + monkeypatch.setattr( + effect_runtime, + "effect_runtime_result", + lambda *args: { + "schema_version": "loopx_state_event_append_result_v0", + "status": "planned", + "choices": [choice], + }, + ) + with pytest.raises(StateEventError, match="invalid event append plan choice"): + store.append(_event("second")) + assert store.path.read_bytes() == before + + +def test_concurrent_processes_publish_contiguous_batches(tmp_path: Path) -> None: + import subprocess + import sys + + event_log = tmp_path / "events.jsonl" + start = tmp_path / "start" + script = """ +import sys, time +from pathlib import Path +from loopx.event_sourced_state import AppendOnlyStateEventStore, make_state_event, TODO_ADDED +path, barrier, worker = sys.argv[1:] +while not Path(barrier).exists(): + time.sleep(0.01) +AppendOnlyStateEventStore(Path(path)).append_many([ + make_state_event(event_id=f"{worker}-{i}", goal_id="goal-a", event_type=TODO_ADDED, + refs={"todo_id": f"todo_{worker}_{i}"}, payload={"role": "agent", "title": "Concurrent batch"}, + recorded_at="2026-09-24T00:00:00Z") for i in range(3) +]) +""" + workers = [ + subprocess.Popen( + [sys.executable, "-c", script, str(event_log), str(start), str(i)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + for i in range(3) + ] + try: + start.touch() + for process in workers: + _, error = process.communicate(timeout=40) + assert process.returncode == 0, error + finally: + for process in workers: + if process.poll() is None: + process.kill() + process.wait() + events = AppendOnlyStateEventStore(event_log).load() + assert [row["append_sequence"] for row in events] == list(range(1, 10)) + for worker in range(3): + positions = [ + i + for i, row in enumerate(events) + if row["event_id"].startswith(f"{worker}-") + ] + assert positions == list(range(positions[0], positions[0] + 3)) From 92ba6b11d8c76aefd9d64b8a58aaea1dbedcdf8a Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Thu, 24 Sep 2026 20:45:17 +0800 Subject: [PATCH 2/2] docs(authority): reconcile event capture and default-cutover path Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- ...2026-09-24-event-completion-transaction.md | 92 +++++++++ ...9-24-event-completion-transaction.zh-CN.md | 72 +++++++ ...shared-goal-authority-state-provider-v0.md | 2 + ...-goal-authority-state-provider-v0.zh-CN.md | 2 + .../typescript-control-plane-migration-v0.md | 2 + ...script-control-plane-migration-v0.zh-CN.md | 2 + .../event-completion-rehearsal.py | 183 ++++++++++++++++++ 7 files changed, 355 insertions(+) create mode 100644 docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.md create mode 100644 docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.zh-CN.md create mode 100644 examples/control_plane/event-completion-rehearsal.py diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.md new file mode 100644 index 0000000000..b869451d2c --- /dev/null +++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.md @@ -0,0 +1,92 @@ +# Event-owned completion: one publication before capture integration + +Baseline: `90f21a5299188d54f984a5313e774c9ac48d6595`. This advances overall +roadmap R5/G2, shared-authority L2/L7 and TS T1/T2. It closes an existing event +writer correctness gap; it does not qualify that writer for shadow capture. + +## Reconcile already-delivered work + +Transaction-bound Markdown/lease shadow outboxes already exist (#3870). +Complete source assembly moved to TS in #4967; prepared-entry source resolution +and delivery moved to TS in #4968. Those owners must be reused. “Event source +capture is missing” was too broad: the remaining gap is binding the event-log +writer's actual commit to the existing capture lifecycle, then qualifying mixed +writers and whole-Goal recovery. Source assembly is not another remaining PR. +The `event_log_writer_not_bound` hold stays in bootstrap, capture and delivery. + +## Behavior and ownership + +Previously event-owned Todo completion appended successor add/claim events +before it finished encoding the parent completion. A later error left runnable +successors without a completed parent. Its context check also did not compare +the actual event log under the append lock. Both failures reproduce on the +baseline through the public completion function. + +- The event adapter now encodes the existing TS successor proposals, then + submits successors and completion as one eager batch. Duplicate normalization + and default/ownership decisions are removed from the Python successor helper. +- `goals/state_event_append.ts` owns whole-batch identity conflicts, replay, + sequence allocation and source-checksum admission. Python holds the existing + sibling lock, supplies compact identity/hash facts and retains legacy codecs + and IO. There is one planning RPC per batch, not per historical event. +- Exact `list`/`tuple` batches validate fully before publication. Atomic replace + plus file/directory fsync makes the event stream visible as all old or all new + bytes. Prior bytes and event schemas remain unchanged. Lazy iterables and + subclasses keep their per-item visibility/reentrancy contract; callers must + materialize them before requesting an atomic source-bound batch. +- Source drift returns the existing completion validation failure, without a + successor prefix. A lost publication acknowledgement reports an uncertain + outcome; read back the original Todo and retry completion. Terminal replay + re-establishes log durability without generating more successors. + +This deliberately changes eager-batch failure/visibility semantics. The log is +logically append-only, but the physical inode is replaced. Consumers must reopen +it; an indefinitely open file descriptor is not a live-tail contract. No event +schema version, capture gate, migration permission or provider default changes. +No frontend setting changes: the public completion result and existing CLI/API +route remain the entrypoints; only failure atomicity and replay are corrected. + +## Evidence and cost + +Public-entrypoint counterexamples fail on baseline and pass on this change. +Validation also covers late duplicate/invalid events, source drift, concurrent +process batches, pre-replace failure, post-replace fsync failure, exact retry, +historical CRLF/no-final-newline preservation, both successor roles and dry-run. +Existing event-only capture holds and non-Todo supervisor/read consumers remain +covered. A source-projection return-value narrowing fixes an existing mypy +failure without changing its runtime acceptance rules. + +The read-only source-copy rehearsal consumed 6,111,476 Markdown bytes and +backfilled 874 events. A disposable registry's real CLI completed a synthetic +Todo with three appended events in 1,531 ms; replay left bytes unchanged and +source digest readback matched. It uses real record variety/volume, not live +Goal configuration, execution or promotion; it does not assert complete archive +capture. `examples/control_plane/event-completion-rehearsal.py` reproduces it. + +On the same 707,414-byte detached log, seven warm three-event batches had median +11.33 ms on baseline and 28.94 ms on this change. This pays for admission and +crash durability; it is not a speedup. Existing whole-log reads remain, and +atomic publication adds a whole-file copy. Do not use this legacy adapter as the +future high-throughput provider; retire it with its final caller after migration. +No RPC budget was raised. File/SQLite/PostgreSQL stores are not changed here. + +## Remaining local-default delivery program + +The conditional estimate remains **5–8 cohesive delivery PRs**, subject to +integration findings and existing open prerequisites. This transaction repair +is a prerequisite within public-writer/capture closure, not grounds to subtract +one complete package. Older 7–9 estimates describe earlier checkpoints. + +| Package | PRs | Concrete exit | +| --- | --- | --- | +| Public callers and executor boundaries | 1–2 | Reconcile real CLI/Turn/Chat writers and external-effect consumers; close current-proof/fence gaps and delete replaced Python rules. Reuse current leased handoff/selection/receipt work. | +| Consumer and display closure (D1) | 1 | Integrate merged #4961 projection recovery, #4964 complete-source summary and #4922 snapshot paging; prove missing/stale display recovery through actual clients. Do not reimplement these owners. | +| Selected SQLite profile (D2) | 1–2 | Continue #4224 and coordinate #4931: capacity, crash/restore/upgrade, lag, supported runtime/OS and applicable elapsed soak on the selected profile. Test count is not elapsed soak. | +| Capture continuity and whole-Goal rehearsal (L7/L8, D3) | 1–2 | Bind actual event transactions to prepared/committed outbox identity; prove mixed writers, interruption, drain, old-writer fencing, canonical readback and fenced rollback on File/SQLite. Keep unbound holds until this passes. | +| Default selection and bounded retirement (L9/T4) | 1 | New-Goal creation/settings/install select the qualified profile; migrate approved existing cohorts and delete old business writers only after final callers and compatibility windows close. Markdown import/export/rendering are not obsolete business writers. | + +File remains the explicit reference profile and SQLite the long-lived local +candidate. PostgreSQL already has a provider and scoped service factory; real +service authentication, deployment, restore/failover and capacity qualification +remain a separate medium-term package. They are not prerequisites for local +default selection and this PR does not qualify them. diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.zh-CN.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.zh-CN.md new file mode 100644 index 0000000000..88b76bea57 --- /dev/null +++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.zh-CN.md @@ -0,0 +1,72 @@ +# 事件源完成事务：先建立完整提交，再接入捕获 + +基线：`90f21a5299188d54f984a5313e774c9ac48d6595`。对应总路线 R5/G2、 +shared-authority L2/L7 与 TS T1/T2。本批修复已有事件写入者的正确性， +不授予该写入者 shadow capture 资格。 + +## 核对已经交付的工作 + +绑定事务的 Markdown/lease shadow outbox 早已存在（#3870）。#4967 已把完整 +来源组装移到 TS，#4968 已把 prepared-entry 的来源判定与交付移到 TS，必须复用。 +笼统地说“事件源捕获还没做”不准确：剩下的是把事件日志写入者的实际提交绑定到 +既有捕获生命周期，再验证混合写入者和整 Goal 恢复。来源组装不应重复列成待做 PR。 +Bootstrap、capture 和 delivery 中的 `event_log_writer_not_bound` 阻塞保持。 + +## 行为与所有权 + +之前，事件源 Todo 完成会先写入后继 add/claim 事件，再编码父任务完成事件。 +后续出错会留下可运行后继，却没有完成父任务。原来的上下文检查也没有在追加锁内 +比较真实事件日志。这两个反例均能通过公开完成函数在基线复现。 + +- 事件适配器现在只编码既有 TS 后继提案，把后继和完成事件作为一个完整批次提交。 + 删除 Python 后继 helper 中重复的归一化、默认值和所有权决定。 +- `goals/state_event_append.ts` 负责整批身份冲突、回放、序号分配和来源校验准入。 + Python 持有既有 sibling lock，传输紧凑身份/hash 事实，保留旧格式 codec 和 IO。 + 每批一次规划 RPC，不逐条历史事件调用。 +- 精确的 `list`/`tuple` 批次先完整验证，再原子替换并 fsync 文件和目录。读者看到 + 全部旧字节或全部新字节；已有字节与事件 schema 不变。惰性迭代器和子类保留逐条 + 可见与可重入合同；请求原子且绑定来源的批次前，调用者须先物化集合。 +- 来源变化返回既有完成校验失败，不留下部分后继。发布确认丢失时报告结果不确定； + 应回读原 Todo 再重试完成。终态重放重新确认日志持久性，不生成更多后继。 + +这是有意改变 eager batch 的失败/可见性语义。日志在逻辑上只追加，但物理 inode +会替换，读者需重新打开；长期持有文件描述符不构成实时尾读合同。不修改事件 schema +版本、capture gate、迁移权限或 provider 默认值。没有前端设置变化：入口仍是既有 +CLI/API 和完成结果，修复的是失败原子性与重放。 + +## 证据与代价 + +公开入口反例在基线失败、在本批通过。覆盖批末重复/非法事件、来源变化、多个进程 +整批竞争、替换前失败、替换后 fsync 失败、精确重试、历史 CRLF/无末尾换行保留、 +两种角色的后继及 dry-run。保留事件独有来源捕获阻塞与非 Todo 的 supervisor/read +消费验证。另将来源投影返回值绑定到经校验的局部变量，修复已有 mypy 失败，运行时 +准入语义不变。 + +只读源副本演练读取 6,111,476 字节 Markdown、回填 874 条事件。在临时 registry +中通过真实 CLI 完成合成 Todo，追加三个事件，耗时 1,531 ms；重试字节不变，真实 +来源 digest 回读一致。使用真实记录多样性/规模，不激活真实 Goal 配置、不执行或 +晋升真实 Goal，也不宣称完整 archive capture。可通过 +`examples/control_plane/event-completion-rehearsal.py` 复现。 + +同一个 707,414 字节隔离日志，七次预热后三事件批次，中位耗时基线 11.33 ms、 +本批 28.94 ms。这是准入与崩溃持久性的成本，不是性能提升。既有整日志读取仍在， +原子发布还增加整文件复制。该旧适配器不是未来高吞吐 provider，应在迁移后随最后 +调用者退役。不提高 RPC 预算，本批不修改 File/SQLite/PostgreSQL store。 + +## 到本地默认切换的剩余交付 + +条件估算仍为 **5–8 个完整交付 PR**，取决于集成发现和已有开放前置 PR。 +本次事务修复属于真实写入者/捕获收口的前置项，不能据此机械减去一个完整包。 +旧的 7–9 批估算对应更早的检查点。 + +| 交付包 | PR 数 | 具体出口 | +| --- | --- | --- | +| 公开 caller 与执行边界 | 1–2 | 核对 CLI/Turn/Chat 真实写入者及外部效果消费者，收口当前证明/fence 缺口，删除被替换的 Python 规则；复用进行中的 leased handoff/selection/receipt 工作。 | +| 消费与显示闭合（D1） | 1 | 集成已合入的 #4961 投影恢复、#4964 完整来源摘要、#4922 快照分页，沿真实客户端证明缺失/过期显示恢复，不重写这些 owner。 | +| 选定 SQLite profile（D2） | 1–2 | 继续 #4224、协调 #4931，完成容量、崩溃/恢复/升级、lag、支持的 runtime/OS 与适用的真实持续运行验证；测试次数不等于持续时间。 | +| 捕获连续性与整 Goal 演练（L7/L8、D3） | 1–2 | 将实际事件事务绑定 prepared/committed outbox 身份；在 File/SQLite 验证混合写入者、中断、排空、旧写入者 fencing、canonical 回读和受 fencing 保护的回滚。通过前保持 unbound 阻塞。 | +| 默认选择与有限退役（L9/T4） | 1 | 新 Goal 创建/设置/安装采用已资格化 profile；迁移获准旧 cohort，最后调用者与兼容窗口关闭后删除旧业务 writer。Markdown import/export/rendering 不属于过时业务 writer。 | + +File 保持显式参考 profile，SQLite 是长期本地候选。PostgreSQL 已有 provider 和 +受作用域约束的 service factory；真实服务认证、部署、恢复/故障切换与容量资格是 +独立中期工作，不应让本地默认切换等它完成，本批也不授予这些资格。 diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md index 187d5d46d4..f333cc77e7 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -38,6 +38,8 @@ are not a guaranteed total PR count. Use the [reconciled inventory and exits](le ## Current implementation checkpoint +The current [event transaction and default-cutover plan](ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.md) estimates 5–8 complete packages conditionally. #4967 source assembly and #4968 capture delivery are already delivered; event-writer binding remains open, with atomic completion repaired here as a prerequisite. Earlier counts below describe historical checkpoints, not additional current work. + Handoff-mode changes now share one TS ownership-fact classifier before and after promotion. Legacy event-only claims reject rather than disappear at a Markdown boundary; event append locks protect the observation through writeback. diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md index b0e7f8226a..e83eacb8bc 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md @@ -34,6 +34,8 @@ ## 当前实现检查点 +当前剩余交付以[事件事务与默认切换计划](ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.zh-CN.md)为准：条件估算 5–8 个完整包。#4967 来源组装、#4968 捕获交付已经完成；事件写入者绑定仍未完成，本批先修复完整完成事务。下文更早的包数属于历史检查点，不能作为当前待办重复计算。 + 终结 caller 现将审核与验证绑定 canonical 来源，历史回执恢复不再依赖私有 argv。 Agent 完成和 Monitor 停止复用普通编辑的当前 head 显示确认。 [调用与恢复合同](../../reference/canonical-terminal-review.zh-CN.md)。此批推进 L2/L5， diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index f7d14f87fe..d6c3104aee 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -29,6 +29,8 @@ Retain T0 caller/parity inventory, T1/T2 transaction/effect convergence, T3 comp ## Current implementation checkpoint +The current [event transaction and default-cutover plan](ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.md) estimates 5–8 complete packages conditionally. #4967 source assembly and #4968 capture delivery are already delivered; event-writer binding remains open, with atomic completion repaired here as a prerequisite. Earlier counts below describe historical checkpoints, not additional current work. + Canonical collection transport now uses snapshot-bound, byte-bounded TS pages. The same `canonicalTodoCollection` owner validates both the retained direct list and paged reads; Python assembles complete pages and preserves the caller shape. diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index abf507c179..53cae59d1f 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -36,6 +36,8 @@ shared-authority 的当前核对表区分已合入实现、在途 PR、新代码 ## 当前实现检查点 +当前剩余交付以[事件事务与默认切换计划](ledger/shared-goal-authority-state-provider-v0/2026-09-24-event-completion-transaction.zh-CN.md)为准：条件估算 5–8 个完整包。#4967 来源组装、#4968 捕获交付已经完成；事件写入者绑定仍未完成，本批先修复完整完成事务。下文更早的包数属于历史检查点，不能作为当前待办重复计算。 + Canonical command 的 receipt/head 观察顺序统一归属 TS：团队规划、Todo 创建/ 修改/领取/终态/归档、Monitor、lease 维护和 Goal acceptance 在读 head 后复查原 receipt，再执行新准入。这修复同 operation 并发竞争，不扩展 provider API、不 diff --git a/examples/control_plane/event-completion-rehearsal.py b/examples/control_plane/event-completion-rehearsal.py new file mode 100644 index 0000000000..6b4cce4f1b --- /dev/null +++ b/examples/control_plane/event-completion-rehearsal.py @@ -0,0 +1,183 @@ +#!/usr/bin/env python3 +"""Exercise event-owned completion on a detached real Markdown snapshot. + +Only synthetic Todos are completed. Source Goal configuration, validators and +runtime are never activated. All writes target a temporary registry and event +log; output contains counts/timings only. This is event-adapter qualification, +not shadow-capture, provider promotion or SQLite soak evidence. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import time + +REPOSITORY = Path(__file__).resolve().parents[2] +if str(REPOSITORY) not in sys.path: + sys.path.insert(0, str(REPOSITORY)) + +from loopx.event_sourced_state import ( # noqa: E402 + TODO_ADDED, + AppendOnlyStateEventStore, + backfill_todo_events_from_markdown, + build_state_projection, + make_state_event, +) +from loopx.history import load_registry # noqa: E402 +from loopx.state_refresh import resolve_goal_state # noqa: E402 + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--registry", type=Path, required=True) + parser.add_argument("--goal-id", required=True) + args = parser.parse_args() + registry_path = args.registry.expanduser().resolve() + registry = load_registry(registry_path) + _, _, source = resolve_goal_state( + registry=registry, + goal_id=args.goal_id, + project_override=None, + state_file_override=None, + ) + source_bytes = source.read_bytes() + source_digest = hashlib.sha256(source_bytes).digest() + # Preserve real record variety and volume, without inheriting a live owner. + events = backfill_todo_events_from_markdown( + source_bytes.decode("utf-8"), + goal_id="event-rehearsal", + recorded_at="2026-09-24T00:00:00Z", + ) + if not events: + raise SystemExit("selected source has no Markdown Todos to rehearse") + with tempfile.TemporaryDirectory(prefix="loopx-event-rehearsal-") as directory: + root = Path(directory) + state = root / "ACTIVE_GOAL_STATE.md" + state.write_text( + "---\ngoal_id: event-rehearsal\n---\n\n## Agent Todo\n", encoding="utf-8" + ) + clone_registry = root / "registry.json" + clone_registry.write_text( + json.dumps( + { + "common_runtime_root": str(root / "runtime"), + "goals": [ + { + "id": "event-rehearsal", + "status": "active", + "repo": str(root), + "state_file": state.name, + "domain": "harness_self_improvement", + "adapter": {"kind": "harness_self_improvement"}, + "coordination": { + "agent_model": "peer_v1", + "registered_agents": ["rehearsal-worker"], + }, + } + ], + } + ), + encoding="utf-8", + ) + store = AppendOnlyStateEventStore(root / "events.jsonl") + store.append_many(events) + initial_count = len(store.load()) + todo_id = "todo_rehearsal_atomic_completion" + assert not any(row.get("refs", {}).get("todo_id") == todo_id for row in events) + store.append( + make_state_event( + event_id="rehearsal-parent", + goal_id="event-rehearsal", + event_type=TODO_ADDED, + refs={"todo_id": todo_id}, + payload={ + "role": "agent", + "title": "Qualify one detached event transaction.", + "task_class": "advancement_task", + "claimed_by": "rehearsal-worker", + }, + recorded_at="2026-09-24T01:00:00Z", + ) + ) + before = store.path.read_bytes() + command = [ + sys.executable, + "-c", + "from loopx.cli import main; raise SystemExit(main())", + "--registry", + str(clone_registry), + "--format", + "json", + "todo", + "complete", + "--goal-id", + "event-rehearsal", + "--todo-id", + todo_id, + "--claimed-by", + "rehearsal-worker", + "--evidence", + "Detached validation passed.", + "--next-agent-todo", + "Independently review the detached delivery.", + "--next-task-class", + "advancement_task", + "--next-claimed-by", + "rehearsal-worker", + "--execute", + ] + started = time.perf_counter() + first = subprocess.run( + command, cwd=REPOSITORY, capture_output=True, text=True, timeout=120 + ) + # Do not include subprocess output: it can contain copied private titles. + if first.returncode: + raise RuntimeError( + f"detached completion CLI failed (exit {first.returncode})" + ) + result = json.loads(first.stdout) + assert result["ok"] and result["source"] == "event_log" + elapsed_ms = round((time.perf_counter() - started) * 1000) + landed = store.path.read_bytes() + assert landed.startswith(before) + projection = build_state_projection(store.load()) + parent = next( + row + for row in projection["agent_todos"]["items"] + if row["todo_id"] == todo_id + ) + assert parent["status"] == "done" and len(parent["successor_todo_ids"]) == 1 + replay = subprocess.run( + command, cwd=REPOSITORY, capture_output=True, text=True, timeout=120 + ) + if replay.returncode: + raise RuntimeError(f"detached replay CLI failed (exit {replay.returncode})") + assert json.loads(replay.stdout)["idempotent_replay"] is True + assert store.path.read_bytes() == landed + assert hashlib.sha256(source.read_bytes()).digest() == source_digest, ( + "live source changed during rehearsal" + ) + print( + json.dumps( + { + "ok": True, + "source_unchanged": True, + "source_bytes": len(source_bytes), + "backfilled_events": initial_count, + "detached_log_bytes": len(landed), + "completion_events_added": len(store.load()) - initial_count - 1, + "cli_completion_ms": elapsed_ms, + "replay_unchanged": True, + } + ) + ) + + +if __name__ == "__main__": + main()