From 843984506c4af48f6cd3dbf686be244a0f921a45 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 20:49:41 -0400 Subject: [PATCH 1/4] Honor registry privacy in notification reads and complete native setup --- src/agent_comms/coordination_schema.py | 9 ++--- src/agent_comms/installer/upgrade.py | 3 +- src/agent_comms/native_runtime_input.py | 22 ++++++++++++ src/agent_comms/notification_assignment.py | 37 ++++++++------------ src/agent_comms/private_nk_entrypoint.py | 1 - src/agent_comms/private_registry_guard.py | 39 +++++++++++++++++++++- src/agent_comms/registration.py | 10 ++++-- src/agent_comms/selected_participant.py | 5 +-- tests/bus_record_fixture.py | 4 --- tests/test_install_announcements.py | 21 ++++++++++++ 10 files changed, 113 insertions(+), 38 deletions(-) diff --git a/src/agent_comms/coordination_schema.py b/src/agent_comms/coordination_schema.py index 3f75f0d083..f9af57f9a2 100644 --- a/src/agent_comms/coordination_schema.py +++ b/src/agent_comms/coordination_schema.py @@ -3,10 +3,10 @@ from __future__ import annotations from functools import cache -from importlib import import_module -from pkgutil import iter_modules from typing import Final +from metaclass_registry.discovery import discover_registry_classes + import agent_comms.coordination_tables as coordination_tables from agent_comms.assignment_states import AssignmentState @@ -42,8 +42,9 @@ def schema_rows(cls): @classmethod def declarations(cls): - for module in iter_modules(coordination_tables.__path__, coordination_tables.__name__ + "."): - import_module(module.name) + discover_registry_classes( + coordination_tables.__path__, coordination_tables.__name__ + ".", CoordinatorTable, + ) return TypedTable.members_with(cls) @classmethod diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py index 329e37789b..49d38c428c 100644 --- a/src/agent_comms/installer/upgrade.py +++ b/src/agent_comms/installer/upgrade.py @@ -689,7 +689,8 @@ def upgrade(checkout: Path, root: Path, target: str, installer: InstallerChoice, bridges.require_resumed() return result except UpgradeRefused as error: - report.failed(error) + if not rehearse_only: + report.failed(error) raise finally: tree.close() diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py index 5a3fd16791..ca94d92ace 100644 --- a/src/agent_comms/native_runtime_input.py +++ b/src/agent_comms/native_runtime_input.py @@ -25,6 +25,7 @@ NativeInputLocation, NativeInputPosition, NativeInputRecord, + TriageNativeExecution, NativeStageStorage, NumberNativeColumn, RecordedNativePosition, @@ -53,6 +54,7 @@ TypedRow, TypedTable, WriteOnceColumns, + sql_literal, ) if TYPE_CHECKING: @@ -99,6 +101,26 @@ class NativeRuntimeInput( NativeRuntimeTable, TypedTable, ): + @classmethod + def notification_context_sql(cls) -> str: + """Native context evidence for the caller's bounded assignment window ``w``.""" + triage_stage = sql_literal(TriageNativeExecution) + return ( + "WITH member AS MATERIALIZED (SELECT m.input_id,m.assignment_id,m.ordinal " + f"FROM ({cls.source_membership_sql()}) m " + "WHERE m.assignment_id IN (SELECT assignment_id FROM w)) " + "SELECT w.assignment_id," + f"EXISTS (SELECT 1 FROM {cls.declared_name} n " + "JOIN member ON member.input_id=n.input_id " + "WHERE member.assignment_id=w.assignment_id " + f"AND n.stage={triage_stage} AND n.session_entry_id IS NULL) AS triage_inflight," + f"(SELECT n.committed_at_ms FROM {cls.declared_name} n " + "JOIN member ON member.input_id=n.input_id " + "WHERE member.assignment_id=w.assignment_id AND n.session_entry_id IS NOT NULL " + f"ORDER BY CASE n.stage WHEN {triage_stage} THEN 0 ELSE 1 END," + "member.ordinal,n.input_id LIMIT 1) AS context_committed_at_ms FROM w" + ) + write_once_columns = ('sent_owner_claimed_generation', 'session_id', 'session_file') finalized_column = 'session_entry_id' required_update_columns = ('sent_owner_claimed_generation',) diff --git a/src/agent_comms/notification_assignment.py b/src/agent_comms/notification_assignment.py index 747e616298..6ee86e03d8 100644 --- a/src/agent_comms/notification_assignment.py +++ b/src/agent_comms/notification_assignment.py @@ -23,9 +23,10 @@ SentProgress, ShownToModelProgress, ) -from .typed_table import BooleanStorage, TypedRow, sql_literal +from .typed_table import BooleanStorage, TypedRow if TYPE_CHECKING: + from .private_registry_guard import RegistryPrivacy from .agent_activity import RecipientActivity from .bus_publication import Delivered from .presentation import MessageNotification @@ -288,7 +289,8 @@ def _recorded_conversation_responses( @classmethod def capture( - cls, connection: sqlite3.Connection, sources: Sequence[Delivered] + cls, connection: sqlite3.Connection, sources: Sequence[Delivered], + privacy: RegistryPrivacy, ) -> NotificationAssignmentRead: """Capture detached assignment evidence for one bounded window of originals from the caller's SQLite snapshot.""" @@ -300,12 +302,8 @@ def capture( if not sources: return NotificationAssignmentRead((), ()) from .coordination_tables.executions import CurrentExecutions - from .native_input_record import TriageNativeExecution - from .native_runtime_input import NativeRuntimeInput - CoordinatorTable.require_schema(connection) columns = ",".join(f"w.{name}" for name in WakeAssignment.columns()) - triage_stage = sql_literal(TriageNativeExecution) # The window's assignments and their native-input membership are each # selected once; membership has no assignment index, so it is built # for these assignments only, not once per row and per column. @@ -313,22 +311,13 @@ def capture( connection.execute( f"WITH w AS MATERIALIZED (SELECT * FROM {WakeAssignment.declared_name} " f"WHERE wire_seq IN ({','.join('?' for _ in sources)})), " - "member AS MATERIALIZED (SELECT m.input_id,m.assignment_id,m.ordinal " - f"FROM ({NativeRuntimeInput.source_membership_sql()}) m " - "WHERE m.assignment_id IN (SELECT assignment_id FROM w)) " - f"SELECT {columns}, EXISTS (SELECT 1 FROM {NativeRuntimeInput.declared_name} n " - "JOIN member ON member.input_id=n.input_id " - "WHERE member.assignment_id=w.assignment_id " - f"AND n.stage={triage_stage} AND n.session_entry_id IS NULL) " - "AS triage_inflight, pointer.execution_id AS current_execution_id, " - f"(SELECT n.committed_at_ms FROM {NativeRuntimeInput.declared_name} n " - "JOIN member ON member.input_id=n.input_id " - "WHERE member.assignment_id=w.assignment_id AND n.session_entry_id IS NOT NULL " - f"ORDER BY CASE n.stage WHEN {triage_stage} THEN 0 ELSE 1 END," - "member.ordinal,n.input_id LIMIT 1) AS context_committed_at_ms, " + f"context AS MATERIALIZED ({privacy.notification_context_sql(connection)}) " + f"SELECT {columns},context.triage_inflight," + "pointer.execution_id AS current_execution_id,context.context_committed_at_ms," "p.seq AS response_seq, p.message_id AS response_message_id, " "p.received_at_ms AS response_received_at_ms " "FROM w " + "JOIN context ON context.assignment_id=w.assignment_id " f"LEFT JOIN {CurrentExecutions.declared_name} pointer " "ON pointer.owner_lookup=w.recipient_lookup " f"LEFT JOIN {PublicationEvidence.declared_name} p " @@ -346,18 +335,20 @@ def read( ) -> NotificationAssignmentRead: """Capture assignment evidence through the single observing connection owner.""" from .coordination_database import CoordinationStore + from .private_registry_guard import RegistryPrivacy + privacy = RegistryPrivacy.capture(root) with CoordinationStore.observed(CoordinationStore.path_in(root)).observing( - cls, sources + cls, sources, privacy, ) as result: return result @classmethod - def missing_store(cls, store, sources): - return NotificationAssignmentRead(tuple(sources), ()) + def missing_store(cls, store, sources, privacy): + return privacy.missing_notification_assignments(sources) @classmethod - def invalid_store(cls, store, sources): + def invalid_store(cls, store, sources, privacy): raise ValueError(f"Channel notification status unavailable: {store.failure_reason}") @property diff --git a/src/agent_comms/private_nk_entrypoint.py b/src/agent_comms/private_nk_entrypoint.py index 35a6f45180..6d4ca303fc 100644 --- a/src/agent_comms/private_nk_entrypoint.py +++ b/src/agent_comms/private_nk_entrypoint.py @@ -81,7 +81,6 @@ def prepare_owner(self, thread: Thread, agent_bin: str, runtime: LaunchRuntime) root = self.child_root Registration(RegistryFile.path_in(root)).prepare_native_root() with Coordination(str(CoordinationStore.path_in(root))) as store: - store.install_private_runtime() store.participants.register( thread.incarnation.require_recorded().lookup, thread.name, thread.name, committed=True, diff --git a/src/agent_comms/private_registry_guard.py b/src/agent_comms/private_registry_guard.py index ae2d5d0ae2..ce40cb40d8 100644 --- a/src/agent_comms/private_registry_guard.py +++ b/src/agent_comms/private_registry_guard.py @@ -19,10 +19,11 @@ import hashlib import json import os +import sqlite3 import stat import struct from abc import ABC, abstractmethod -from collections.abc import Iterable +from collections.abc import Iterable, Sequence from dataclasses import dataclass, replace from pathlib import Path from typing import TYPE_CHECKING, Annotated @@ -33,9 +34,12 @@ from .errors import RelationViolationError from .field_codec import FieldCodec, Id128Text, MalformedRecord, WireName from .locked_store import DurableFile, WireWakeFile +from .native_runtime_input import NativeRuntimeInput, NativeRuntimeTable from .private_path import fsync_directory if TYPE_CHECKING: + from .bus_publication import Delivered + from .notification_assignment import NotificationAssignmentRead from .acp_extension import CursorAdvancedUpdate, CursorScope from .cursor_publication import CursorPublication from .private_nk_entrypoint import RootLaunch @@ -155,8 +159,30 @@ def prepare_native(self, registry_path: Path) -> PrivateRegistryGuard: ... @abstractmethod def require_root(self, root_id: str, error: Exception) -> None: ... + @abstractmethod + def notification_context_sql(self, connection: sqlite3.Connection) -> str: + """Context evidence for every row in the captured assignment window ``w``.""" + + @abstractmethod + def missing_notification_assignments( + self, sources: Sequence[Delivered], + ) -> NotificationAssignmentRead: ... + class PublicRegistry(RegistryPrivacy): + def notification_context_sql(self, connection: sqlite3.Connection) -> str: + return ( + "SELECT assignment_id,0 AS triage_inflight," + "NULL AS context_committed_at_ms FROM w" + ) + + def missing_notification_assignments( + self, sources: Sequence[Delivered], + ) -> NotificationAssignmentRead: + from .notification_assignment import NotificationAssignmentRead + + return NotificationAssignmentRead(tuple(sources), ()) + def cursor_scope( self, snapshot: RegistrySnapshot, thread_name: str, session_id: str, ) -> CursorScope: @@ -197,6 +223,17 @@ def require_root(self, root_id: str, error: Exception) -> None: class PrivateRegistryGuard(DurableFile, WireWakeFile, RegistryPrivacy): filename = ".registry-owner-guard" + def notification_context_sql(self, connection: sqlite3.Connection) -> str: + NativeRuntimeTable.require_installed(connection) + return NativeRuntimeInput.notification_context_sql() + + def missing_notification_assignments( + self, sources: Sequence[Delivered], + ) -> NotificationAssignmentRead: + from .coordination_errors import PublicationActivationBlocked + + raise PublicationActivationBlocked("Private notification coordinator is missing") + def __init__(self, registry_path: Path, wire_root_id: str): try: Id128Text.decode(wire_root_id) diff --git a/src/agent_comms/registration.py b/src/agent_comms/registration.py index 8dfba86c3e..05875ab153 100644 --- a/src/agent_comms/registration.py +++ b/src/agent_comms/registration.py @@ -86,10 +86,16 @@ def register( self._commit_registration(edit, change) def prepare_native_root(self) -> PrivateRegistryGuard: - """Explicit owner launch prepares registry authority before any coordinator exists.""" + """Prepare native authority, then its complete runtime outside the registry lock.""" + from .coordination_database import CoordinationStore + from .coordinator import Coordination + with self.store.locked(): self.store.read_snapshot_unlocked().purpose.require_write() - return self.store.private_guard_unlocked().prepare_native(self.store.path) + guard = self.store.private_guard_unlocked().prepare_native(self.store.path) + with Coordination(CoordinationStore.path_in(self.store.path.parent)) as store: + store.install_private_runtime() + return guard def register_commander_child(self, name: str, parent_name: str, goals: Goals) -> Thread: """Declare or resume one child and its own request under the registry's lock.""" diff --git a/src/agent_comms/selected_participant.py b/src/agent_comms/selected_participant.py index 6a88bcbdff..a689fe7257 100644 --- a/src/agent_comms/selected_participant.py +++ b/src/agent_comms/selected_participant.py @@ -222,7 +222,8 @@ def prepare( """Capture detached registry, participant and optional context in one worker.""" with comms.registry.store.locked(mode=SharedLock()): _require_no_private_owner_rename(comms.root) - comms.registry.store.private_guard_unlocked().require_root( + privacy = comms.registry.store.private_guard_unlocked() + privacy.require_root( root_id, IdentityConflict("Selected claim namespace changed")) try: thread, generation = comms.registry.live_owner_with_claim(name) @@ -249,7 +250,7 @@ def prepare( store, lookup, thread.name, after_seq=after_seq ) sent_read = NotificationAssignment.capture( - store.session._connection, sent_sources + store.session._connection, sent_sources, privacy, ) if pending: participant.pointer.require_idle() diff --git a/tests/bus_record_fixture.py b/tests/bus_record_fixture.py index 1784746920..adf03269ff 100644 --- a/tests/bus_record_fixture.py +++ b/tests/bus_record_fixture.py @@ -7,8 +7,6 @@ from agent_comms.audience_manifest import FrozenRecipient, freeze_audience from agent_comms.bus import Bus, PrivateRecord from agent_comms.comms import Comms -from agent_comms.coordination_database import CoordinationStore -from agent_comms.coordinator import Coordination from agent_comms.child_process import ProcessIdentity from agent_comms.bus_publication import Delivered, KeyedPublication from agent_comms.envelope_claim_transitions import UnscopedClaims @@ -27,8 +25,6 @@ def prepare_native_comms(comms: Comms) -> Comms: """Validate native-root authority before installing data or opening a coordinator.""" comms.registry.prepare_native_root() Bus.install(Bus.path_in(comms.root)) - with Coordination(str(CoordinationStore.path_in(comms.root))) as coordinator: - coordinator.install_private_runtime() return comms diff --git a/tests/test_install_announcements.py b/tests/test_install_announcements.py index 40cdbb5e2b..5bbfd8cb23 100644 --- a/tests/test_install_announcements.py +++ b/tests/test_install_announcements.py @@ -30,15 +30,36 @@ UnchangedRun, UnresumedUpgrade, Upgrade, + upgrade, ) +from agent_comms.install_choices import UnnamedInstaller from agent_comms.launch_runtime import LaunchRuntime from agent_comms.thread_execution import ExternalThreadExecution from conftest import declare_test_bus +from source_revision_fixture import SourceRepository RESTARTED = RestartedService("Connector on port 1 for /bus", Path("/runtime/bin/python")) +def test_failed_rehearsal_does_not_publish_to_the_source_bus(private_comms, tmp_path): + source = SourceRepository.create(tmp_path / "source") + target = source.git(source.checkout, "rev-parse", "HEAD").strip() + (source.checkout / "owner.py").write_text("# Tracked local change refuses the rehearsal.\n") + before_bus = private_comms.bus.bus.revision() + before_registry = private_comms.registry.snapshot() + + with pytest.raises(UpgradeRefused, match="local changes"): + upgrade( + source.checkout, private_comms.root, target, UnnamedInstaller(), True, False, + tmp_path / "spill", tmp_path / "scratch", lambda: None, + InstallReport(private_comms.root), + ) + + assert private_comms.bus.bus.revision() == before_bus + assert private_comms.registry.snapshot() == before_registry + + @pytest.mark.parametrize("live", ( CarriedRun( steps=(StepRun("carry_order_scopes", Carried(6)),), From 5cd2e45eeb8ac75a568855e2fca469d5c818909c Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 21:01:09 -0400 Subject: [PATCH 2/4] Derive coordinator carry from checked registry privacy --- src/agent_comms/bus_tables.py | 15 +----- src/agent_comms/cli_commands.py | 13 +++-- src/agent_comms/cohort_schema.py | 54 ++++++------------- src/agent_comms/coordinated_runtime_schema.py | 15 ------ src/agent_comms/coordination_database.py | 7 +++ src/agent_comms/coordination_response.py | 16 +----- src/agent_comms/coordinator.py | 14 ----- src/agent_comms/native_prompt_binding.py | 13 +---- src/agent_comms/native_runtime_input.py | 17 +----- src/agent_comms/private_registry_guard.py | 25 +++++++++ src/agent_comms/private_runtime_schema.py | 45 +++++++++++++--- src/agent_comms/registration.py | 6 +-- src/agent_comms/registry_store.py | 8 ++- src/agent_comms/typed_table.py | 26 ++++++++- tests/test_acp_private_nk_delivery.py | 4 +- tests/test_acp_selected_write_plan.py | 13 ++--- tests/test_busy_history_loading.py | 3 +- tests/test_certified_bootstrap.py | 6 +-- tests/test_claim_admission_verifier.py | 4 +- tests/test_claim_bus_read.py | 4 +- tests/test_cohort_schema.py | 35 ++++++------ tests/test_coordinated_runtime.py | 6 +-- tests/test_coordination_cohort.py | 7 ++- tests/test_coordination_response.py | 13 ++--- tests/test_declarations.py | 3 +- tests/test_install_announcements.py | 3 +- tests/test_native_evidence_custody.py | 7 +-- tests/test_optional_awareness_projection.py | 9 ++-- tests/test_private_human_ingress.py | 4 +- tests/test_reconcile_stores.py | 51 ++++++++++++++++++ tests/test_roster_restoration.py | 4 +- tests/test_runtime.py | 3 +- tests/test_selected_answer_forms.py | 4 +- tests/test_selected_tool_broker.py | 4 +- tests/test_tool_refusal_reasons.py | 3 +- tests/test_transcript_partial_tail.py | 3 +- 36 files changed, 251 insertions(+), 216 deletions(-) delete mode 100644 src/agent_comms/coordinated_runtime_schema.py diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py index 617b6ecfdf..0e3b5b6c03 100644 --- a/src/agent_comms/bus_tables.py +++ b/src/agent_comms/bus_tables.py @@ -26,6 +26,7 @@ from agent_comms.typed_table import ( AppendOnlyTables, Column, + ComposedSchemas, DerivedTables, DurableTables, ForeignKey, @@ -57,21 +58,9 @@ def create_schema(cls, db: sqlite3.Connection) -> None: header.kind.read_from(db, header.seq).project_into(db) -class BusSchema(BusTables, BusProjectionTables): +class BusSchema(ComposedSchemas, BusTables, BusProjectionTables): """One file reconciles its declared families using each family's carry policy.""" - @classmethod - def schema_rows(cls) -> tuple[type[TypedTable], ...]: - return tuple(dict.fromkeys( - row for family in cls.__bases__ for row in family.schema_rows() - )) - - @classmethod - def carry_schema(cls, db: sqlite3.Connection) -> None: - for family in cls.__bases__: - family.carry_schema(db) - - @dataclass(frozen=True) class ParticipantSource(TypedTable, BusProjectionTables, declared_name=None): lookup: Annotated[str, Column(primary_key=True)] diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py index 12697ba290..d456bde03f 100644 --- a/src/agent_comms/cli_commands.py +++ b/src/agent_comms/cli_commands.py @@ -1812,6 +1812,10 @@ def answer(self, root: Path) -> ReconciledStoresResult: def apply(self, ctx: Comms) -> ReconciledStoresResult: return self.answer(ctx.root) + def execute(self, route: CommsRoute, output: TextIO) -> int: + """Reconciliation reads its declared files before opening runtime consumers.""" + return self.written(self.answer(route.root), output) + @dataclass(frozen=True, kw_only=True) class ReconcileStoresCliCommand(StoreReconcileCliCommand, declared_name="reconcile-stores"): @@ -1830,14 +1834,9 @@ class PlannedReconcileCliCommand(StoreReconcileCliCommand, declared_name="planne @staticmethod def changes(owner: type[ReconciledFile], path: Path) -> bool: + schema = owner.schema_at(path) with closing(owner.read_snapshot(path)) as db: - return owner.table_schema().schema_differs(db) - - def execute(self, route: CommsRoute, output: TextIO) -> int: - """Opening the bus opens stores, and opening a store reconciles its schema: before - the backup that write would have no copy, so only the stores themselves are read.""" - return self.written(self.answer(route.root), output) - + return schema.schema_differs(db) @dataclass(frozen=True, kw_only=True) class CheckpointCliCommand(NameArgument, CliCommand, declared_name="checkpoint"): diff --git a/src/agent_comms/cohort_schema.py b/src/agent_comms/cohort_schema.py index a33d731572..0d69447ebc 100644 --- a/src/agent_comms/cohort_schema.py +++ b/src/agent_comms/cohort_schema.py @@ -9,11 +9,9 @@ from typing import TYPE_CHECKING, Annotated, ClassVar, Literal from agent_comms.coordination_errors import IdentityConflict, SchemaVersionError -from agent_comms.coordination_schema import CoordinatorTable from agent_comms.coordination_tables.assignments import WakeAssignment from agent_comms.coordination_tables.participants import OwnerGenerations, OwnerRecord, Participants -from agent_comms.coordinator import Coordination -from agent_comms.private_runtime_schema import PrivateRuntimeSchema +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from .audience_manifest import MAX_AUDIENCE_BYTES from .coordination_contracts import POLICY_VERSION, RESOLVER_VERSION @@ -33,7 +31,6 @@ InitialOneWayColumn, NoDelete, SharedFileTables, - SQLiteJournalMode, TypedTable, ) @@ -44,16 +41,12 @@ _LOG = logging.getLogger(__name__) -class CohortTable(PrivateRuntimeSchema): +class CohortTable(PrivateCoordinatorTable): """Mandatory sealed N/K acknowledgement declarations.""" __slots__ = () schema_error = SchemaVersionError - @classmethod - def install(cls, store) -> None: - install_private_cohort_schema(store) - class AwarenessTable(SharedFileTables, DurableTables): """Optional provenance captured only during the unsealed acceptance transaction.""" @@ -61,6 +54,20 @@ class AwarenessTable(SharedFileTables, DurableTables): __slots__ = () schema_error = SchemaVersionError + @classmethod + def install_optional(cls, db: sqlite3.Connection) -> None: + """Optional provenance cannot roll back mandatory native preparation.""" + db.execute("SAVEPOINT optional_awareness_install") + try: + if not cls.stored_schema(db): + cls.create_schema(db) + assert_optional_awareness_schema(db) + except (sqlite3.Error, SchemaVersionError, MalformedRecord) as error: + db.execute("ROLLBACK TO optional_awareness_install") + _LOG.warning("Optional awareness schema unavailable: %r", error, exc_info=error) + finally: + db.execute("RELEASE optional_awareness_install") + @dataclass(frozen=True, kw_only=True) class ClaimBatchAcknowledgements( @@ -366,32 +373,3 @@ def references(cls): def assert_optional_awareness_schema(db: sqlite3.Connection) -> None: if not AwarenessTable.stored_schema(db) or AwarenessTable.schema_differs(db): raise SchemaVersionError("optional awareness generation schema is unavailable") - - -def _install_optional_awareness_schema(db: sqlite3.Connection) -> None: - if not AwarenessTable.stored_schema(db): - AwarenessTable.create_schema(db) - assert_optional_awareness_schema(db) - - -def install_private_cohort_schema(store: Coordination) -> None: - """Create current acknowledgement tables on a fresh coordinator; no old-schema repair.""" - with store.session.writing() as db: - CoordinatorTable.require_schema(db) - if not CohortTable.stored_schema(db): - CohortTable.create_schema(db) - CohortTable.require_installed(db) - if SQLiteJournalMode.read(db.execute("PRAGMA journal_mode")) != [ - SQLiteJournalMode("delete") - ]: - raise SchemaVersionError("cohort schema requires rollback-journal mode") - # Missing/corrupt optional metadata must not roll back mandatory cohort - # creation, and no prior claim is retroactively assigned a generation. - db.execute("SAVEPOINT optional_awareness_install") - try: - _install_optional_awareness_schema(db) - except (sqlite3.Error, SchemaVersionError, MalformedRecord) as error: - db.execute("ROLLBACK TO optional_awareness_install") - _LOG.warning("Optional awareness schema unavailable: %r", error, exc_info=error) - finally: - db.execute("RELEASE optional_awareness_install") diff --git a/src/agent_comms/coordinated_runtime_schema.py b/src/agent_comms/coordinated_runtime_schema.py deleted file mode 100644 index 35ff34b862..0000000000 --- a/src/agent_comms/coordinated_runtime_schema.py +++ /dev/null @@ -1,15 +0,0 @@ -"""Install and verify the one current declaration-derived native schema.""" - -from __future__ import annotations - -from agent_comms.coordinator import Coordination - -from .native_runtime_input import NativeRuntimeTable - - -def install_native_runtime_schema(store: Coordination) -> None: - """Explicit fresh schema install. Existing state must match the declared schema.""" - with store.session.writing() as db: - if not NativeRuntimeTable.stored_schema(db): - NativeRuntimeTable.create_schema(db) - NativeRuntimeTable.require_installed(db) diff --git a/src/agent_comms/coordination_database.py b/src/agent_comms/coordination_database.py index 5ef8f8e5d2..7a6e8ce07e 100644 --- a/src/agent_comms/coordination_database.py +++ b/src/agent_comms/coordination_database.py @@ -18,6 +18,7 @@ CarriedWriter, ReconciledFile, StoreWriter, + TableSchema, ) from .declared_family import DeclaredFamily @@ -72,6 +73,12 @@ class CoordinationStore(CarriedWriter, ReconciledFile): def table_schema(cls) -> type[CoordinatorTable]: return CoordinatorTable + @classmethod + def schema_at(cls, path: Path) -> type[TableSchema]: + from .private_registry_guard import RegistryPrivacy + + return RegistryPrivacy.capture(path.parent).coordinator_schema() + @staticmethod def observed(path: Path) -> StoreFile: """Classify through SQLite without dropping another connection's process locks. diff --git a/src/agent_comms/coordination_response.py b/src/agent_comms/coordination_response.py index 1d1482e809..2da5b96167 100644 --- a/src/agent_comms/coordination_response.py +++ b/src/agent_comms/coordination_response.py @@ -42,7 +42,7 @@ from agent_comms.native_input_owner import RegistryOwner from agent_comms.obligation_states import PublishedResponse, PublishingResponse from agent_comms.owner_fence import OwnerFence -from agent_comms.private_runtime_schema import PrivateRuntimeSchema +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from agent_comms.private_registry_guard import RegistryPrivacy from agent_comms.private_send_rules import RunningTurnRegistryCheck from agent_comms.publisher import ResponseWrites @@ -59,15 +59,11 @@ from .locked_store import RegistryFile -class ResponseTable(ImmutableRows, PrivateRuntimeSchema): +class ResponseTable(ImmutableRows, PrivateCoordinatorTable): """An immutable response definition with its write checks and append evidence.""" schema_error = PublicationActivationBlocked - @classmethod - def install(cls, store) -> None: - install_private_response_schema(store) - @dataclass(frozen=True) class PublicationAppendDispatches(ResponseTable, TypedTable): execution_id: Annotated[str, Column(primary_key=True)] @@ -224,14 +220,6 @@ def require_source(self, initial, assignment: WakeAssignment) -> None: raise IdentityConflict("response sealed route differs from committed source") -def install_private_response_schema(store: Coordination) -> None: - """Install the current response tables on a fresh coordinator only.""" - with store.session.writing() as db: - if not ResponseTable.stored_schema(db): - ResponseTable.create_schema(db) - ResponseTable.require_installed(db) - - @dataclass(frozen=True, kw_only=True) class LiveResponseOwner(RegistryOwner): """One captured registry owner; response finality does not depend on its goal.""" diff --git a/src/agent_comms/coordinator.py b/src/agent_comms/coordinator.py index 7b85d8e789..e1f822b3bb 100644 --- a/src/agent_comms/coordinator.py +++ b/src/agent_comms/coordinator.py @@ -25,7 +25,6 @@ ) from .execution_store import ExecutionStore from .participant_store import ParticipantStore -from .private_runtime_schema import PrivateRuntimeSchema from .store_files import StoreLock from .working_memory_annotations import WorkingMemoryAnnotations @@ -206,19 +205,6 @@ def _bind_tables(self) -> None: self.assignments = AssignmentStore(self.session) self.annotations = WorkingMemoryAnnotations(self.session) - def install_private_runtime(self) -> None: - """Explicit protocol/owner bootstrap, never invoked by a reader.""" - # Load the canonical declarations before querying their existing family. - from agent_comms import ( # noqa: F401 - cohort_schema, - coordination_response, - native_prompt_binding, - native_runtime_input, - ) - - for schema in PrivateRuntimeSchema.__subclasses__(): - schema.install(self) - def close(self) -> None: self.session.close() diff --git a/src/agent_comms/native_prompt_binding.py b/src/agent_comms/native_prompt_binding.py index 9a35c67369..92ccba6312 100644 --- a/src/agent_comms/native_prompt_binding.py +++ b/src/agent_comms/native_prompt_binding.py @@ -18,7 +18,6 @@ from agent_comms.coordination_errors import IdentityConflict from agent_comms.coordination_tables.assignments import WakeAssignment from agent_comms.coordinator import Coordination -from agent_comms.private_runtime_schema import PrivateRuntimeSchema from .cohort_schema import CohortTable from .field_codec import FieldCodec, Id128Text, MalformedRecord @@ -34,6 +33,7 @@ from .threads import Thread from .typed_table import ( Column, + DurableTables, ImmutableRows, IntegerStorage, TextStorage, @@ -52,14 +52,10 @@ def binding_store_path(store: Coordination) -> Path: @dataclass(frozen=True, slots=True) class PromptBinding( - ImmutableRows, DurableFile, SQLiteFile, NativeInputRecord, TypedTable, PrivateRuntimeSchema + ImmutableRows, DurableFile, SQLiteFile, NativeInputRecord, TypedTable, DurableTables ): filename = "native_prompt_bindings.sqlite3" - @classmethod - def install(cls, store) -> None: - install_prompt_binding_schema(store) - input_id: Annotated[ str, Column(primary_key=True, check="length(input_id)=32 AND input_id NOT GLOB '*[^0-9a-f]*'"), @@ -99,11 +95,6 @@ class _BindingRoot(TypedRow): wire_root_id: str -def install_prompt_binding_schema(store: Coordination) -> None: - """Explicit fresh-root install; ordinary callers may rely on _ensure.""" - _ensure_binding_schema(store) - - def _ensure_binding_schema(store: Coordination) -> None: """Serialized snapshot installation; never repair an uncertain commit.""" create_sidecar_file(binding_store_path(store), PromptBinding) diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py index ca94d92ace..33d696970e 100644 --- a/src/agent_comms/native_runtime_input.py +++ b/src/agent_comms/native_runtime_input.py @@ -13,7 +13,7 @@ from agent_comms.coordination_tables.executions import ExecutionRecord from agent_comms.coordination_tables.participants import Participants -from agent_comms.private_runtime_schema import PrivateRuntimeSchema +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from .coordination_errors import IdentityConflict, PublicationActivationBlocked, StaleFence from .field_codec import FieldCodec, MalformedRecord, NativeContext, WireName @@ -62,25 +62,12 @@ from .native_session_reopen import NativeSessionIdentity -class NativeRuntimeTable(PrivateRuntimeSchema): +class NativeRuntimeTable(PrivateCoordinatorTable): """Capability identifying the native-runtime schema's declared tables.""" __slots__ = () schema_error = PublicationActivationBlocked - @classmethod - def install(cls, store) -> None: - from .coordinated_runtime_schema import install_native_runtime_schema - - install_native_runtime_schema(store) - - @classmethod - def schema_rows(cls) -> tuple[type[TypedTable], ...]: - # Importing it declares the family's last table. - from . import triage_native_sources # noqa: F401 - - return super().schema_rows() - @dataclass(frozen=True) class PublishedReplyRevision(TypedRow): diff --git a/src/agent_comms/private_registry_guard.py b/src/agent_comms/private_registry_guard.py index ce40cb40d8..296230f795 100644 --- a/src/agent_comms/private_registry_guard.py +++ b/src/agent_comms/private_registry_guard.py @@ -38,6 +38,7 @@ from .private_path import fsync_directory if TYPE_CHECKING: + from .typed_table import TableSchema from .bus_publication import Delivered from .notification_assignment import NotificationAssignmentRead from .acp_extension import CursorAdvancedUpdate, CursorScope @@ -168,8 +169,19 @@ def missing_notification_assignments( self, sources: Sequence[Delivered], ) -> NotificationAssignmentRead: ... + @abstractmethod + def coordinator_schema(self) -> type[TableSchema]: + """The complete schema this root's authority requires during reconciliation.""" + + class PublicRegistry(RegistryPrivacy): + def coordinator_schema(self) -> type[TableSchema]: + from .coordination_schema import CoordinatorTable + + CoordinatorTable.declarations() + return CoordinatorTable + def notification_context_sql(self, connection: sqlite3.Connection) -> str: return ( "SELECT assignment_id,0 AS triage_inflight," @@ -223,6 +235,19 @@ def require_root(self, root_id: str, error: Exception) -> None: class PrivateRegistryGuard(DurableFile, WireWakeFile, RegistryPrivacy): filename = ".registry-owner-guard" + def coordinator_schema(self) -> type[TableSchema]: + from .private_runtime_schema import PrivateCoordinatorSchema, PrivateCoordinatorTable + + PrivateCoordinatorTable.declarations() + return PrivateCoordinatorSchema + + def prepare_runtime(self) -> None: + from .private_runtime_schema import PrivateCoordinatorTable + + PrivateCoordinatorTable.declarations() + with Coordination(CoordinationStore.path_in(self.registry_path.parent)) as store: + PrivateCoordinatorTable.install(store) + def notification_context_sql(self, connection: sqlite3.Connection) -> str: NativeRuntimeTable.require_installed(connection) return NativeRuntimeInput.notification_context_sql() diff --git a/src/agent_comms/private_runtime_schema.py b/src/agent_comms/private_runtime_schema.py index 94614e289d..ac10168c93 100644 --- a/src/agent_comms/private_runtime_schema.py +++ b/src/agent_comms/private_runtime_schema.py @@ -1,16 +1,49 @@ -"""Explicit private-runtime installation capability for schema declarations.""" +"""Private coordinator declarations and the complete target's derived schema view.""" from __future__ import annotations -from abc import abstractmethod +from .coordination_errors import SchemaVersionError +from .coordination_schema import CoordinatorTable +from .typed_table import ComposedSchemas, DurableTables, SharedFileTables, SQLiteJournalMode, TypedTable -from .typed_table import DurableTables, SharedFileTables +class PrivateCoordinatorTable(SharedFileTables, DurableTables): + """The mandatory private capabilities that share the coordinator file.""" -class PrivateRuntimeSchema(SharedFileTables, DurableTables): - """Durable private tables own both installation and stored-state discovery.""" + schema_error = SchemaVersionError + + @classmethod + def declarations(cls): + from . import ( # noqa: F401 - declarations register in the existing TypedTable family + cohort_schema, + coordination_response, + native_runtime_input, + triage_native_sources, + ) + + return TypedTable.members_with(cls) + + @classmethod + def schema_rows(cls): + return cls.declarations() @classmethod - @abstractmethod def install(cls, store) -> None: """Install absent current state; reject drift without repairing it.""" + cls.declarations() + from .cohort_schema import AwarenessTable + + with store.session.writing() as db: + CoordinatorTable.require_schema(db) + if not cls.stored_schema(db): + cls.create_schema(db) + cls.require_installed(db) + if SQLiteJournalMode.read(db.execute("PRAGMA journal_mode")) != [ + SQLiteJournalMode("delete") + ]: + raise SchemaVersionError("private coordinator requires rollback-journal mode") + AwarenessTable.install_optional(db) + + +class PrivateCoordinatorSchema(ComposedSchemas, CoordinatorTable, PrivateCoordinatorTable): + """The private file target, derived from its core and private row families.""" diff --git a/src/agent_comms/registration.py b/src/agent_comms/registration.py index 05875ab153..061b5d0a3c 100644 --- a/src/agent_comms/registration.py +++ b/src/agent_comms/registration.py @@ -87,14 +87,10 @@ def register( def prepare_native_root(self) -> PrivateRegistryGuard: """Prepare native authority, then its complete runtime outside the registry lock.""" - from .coordination_database import CoordinationStore - from .coordinator import Coordination - with self.store.locked(): self.store.read_snapshot_unlocked().purpose.require_write() guard = self.store.private_guard_unlocked().prepare_native(self.store.path) - with Coordination(CoordinationStore.path_in(self.store.path.parent)) as store: - store.install_private_runtime() + guard.prepare_runtime() return guard def register_commander_child(self, name: str, parent_name: str, goals: Goals) -> Thread: diff --git a/src/agent_comms/registry_store.py b/src/agent_comms/registry_store.py index d8a616fe83..4c12d05743 100644 --- a/src/agent_comms/registry_store.py +++ b/src/agent_comms/registry_store.py @@ -233,6 +233,10 @@ class RegistryStore(RegistryFile, CarriedDocument[RegistryDocument]): cache: RegistryCache = field(init=False, compare=False, repr=False) def __post_init__(self) -> None: + from .private_runtime_schema import PrivateCoordinatorTable + + # Guard-loss checks must know every private declaration before taking this store's lock. + PrivateCoordinatorTable.declarations() object.__setattr__(self, "cache", RegistryCache.for_path(self.path)) record_type = RegistryDocument @@ -403,10 +407,10 @@ def _check_guard_unlocked(self) -> RegistryPrivacy: return PrivateRegistryGuard.load(self.path) coordinator = CoordinationStore.path_in(self.path.parent) if coordinator.exists() or coordinator.is_symlink(): - from .private_runtime_schema import PrivateRuntimeSchema + from .private_runtime_schema import PrivateCoordinatorTable with CoordinationStore.observing(coordinator) as db: - if PrivateRuntimeSchema.stored_schema(db): + if PrivateCoordinatorTable.stored_schema(db): raise RelationViolationError("Native runtime has no registry authority") return PublicRegistry() diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index 0736c4a702..a60fb51395 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -842,6 +842,25 @@ def carry_schema(cls, db: sqlite3.Connection) -> None: """Inside the reconcile's transaction: this kind of file's way onto its declaration.""" +class ComposedSchemas: + """One file derives its rows and carry policy from its nominal schema parents.""" + + @classmethod + def schema_families(cls) -> tuple[type[TableSchema], ...]: + return tuple(family for family in cls.__bases__ if issubclass(family, TableSchema)) + + @classmethod + def schema_rows(cls) -> tuple[type[TypedTable], ...]: + return tuple(dict.fromkeys( + row for family in cls.schema_families() for row in family.schema_rows() + )) + + @classmethod + def carry_schema(cls, db: sqlite3.Connection) -> None: + for family in cls.schema_families(): + family.carry_schema(db) + + class RowMutation(Enum): """SQLite's destructive row events own their immutable-history guards.""" @@ -1793,6 +1812,10 @@ class ReconciledFile(SQLiteFile): def table_schema(cls) -> type[TableSchema]: """The declaration this file's schema is reconciled onto.""" + @classmethod + def schema_at(cls, path: Path) -> type[TableSchema]: + return cls.table_schema() + @classmethod def schema_change(cls, path: Path) -> SchemaChange: return HeldBusFile(cls, path) @@ -1800,8 +1823,9 @@ def schema_change(cls, path: Path) -> SchemaChange: @classmethod def reconcile_at(cls, path: Path) -> bool: """Whether reconciling the file at ``path`` changed it.""" + schema = cls.schema_at(path) with _store_lock(path), closing(cls.writer(path, isolation_level=None)) as db: - return cls.table_schema().reconcile_schema(db, cls.schema_change(path)) + return schema.reconcile_schema(db, cls.schema_change(path)) class CarriedWriter(StoreWriter, DurableFile): diff --git a/tests/test_acp_private_nk_delivery.py b/tests/test_acp_private_nk_delivery.py index a514fec6b5..0fd3e2b835 100644 --- a/tests/test_acp_private_nk_delivery.py +++ b/tests/test_acp_private_nk_delivery.py @@ -6,6 +6,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from bus_record_fixture import prepare_native_comms @@ -23,7 +24,6 @@ from agent_comms.comms import Comms from agent_comms.coordination_errors import ParticipantNotRegistered, PublicationActivationBlocked, StaleRevision from agent_comms.coordinator import Coordination -from agent_comms.coordinated_runtime_schema import install_native_runtime_schema from agent_comms.errors import RelationViolationError from test_coordinated_runtime import tmp_path as private_root_fixture from agent_comms.thread_model import DeclaredModel @@ -44,7 +44,7 @@ def _private_comms(tmp_path): store.participants.register( RecordedThreadIncarnation("fixture", owner.created_at).lookup, "beta", "beta", committed=True ) - install_native_runtime_schema(store) + PrivateCoordinatorTable.install(store) return comms, root_id diff --git a/tests/test_acp_selected_write_plan.py b/tests/test_acp_selected_write_plan.py index a9663f8a92..9086fbd32c 100644 --- a/tests/test_acp_selected_write_plan.py +++ b/tests/test_acp_selected_write_plan.py @@ -2,6 +2,8 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable +from bus_record_fixture import prepare_native_comms from conftest import responder import asyncio @@ -23,15 +25,11 @@ ) from agent_comms.child_process import ProcessIdentity from agent_comms.cohort_foreground import _accept_visible_deliveries -from agent_comms.cohort_schema import install_private_cohort_schema from agent_comms.comms import Comms from agent_comms.launch_root import LaunchRoot -from agent_comms.coordinated_runtime_schema import install_native_runtime_schema from agent_comms.coordination_cohort import sealed_cohort_assignments from agent_comms.envelope_claim_transitions import UnclaimedResource -from agent_comms.coordination_response import install_private_response_schema from agent_comms.coordinator import Coordination -from agent_comms.native_prompt_binding import install_prompt_binding_schema from agent_comms.native_runtime_input import NativeRuntimeInput from agent_comms.thread_model import DeclaredModel from native_backend_fixture import native_backend_fixture @@ -49,7 +47,7 @@ async def test_second_pid_public_acp_owner_ipc_preplan(tmp_path, native_runtime_ root, work = native.root, native.project resource = work / "module.py" resource.write_bytes(b"before\n") - comms = Comms(root) + comms = prepare_native_comms(Comms(root)) sender_pid = os.getpid() comms.registry.declare( comms.registry.new_thread('sender', frozenset(), str(work), process_identity=ProcessIdentity.capture(sender_pid), created_at=61001.0) @@ -59,10 +57,7 @@ async def test_second_pid_public_acp_owner_ipc_preplan(tmp_path, native_runtime_ ) root_id = comms.owners.require_private_launch().wire_root_id with Coordination(str(root / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) - install_private_response_schema(store) - install_native_runtime_schema(store) - install_prompt_binding_schema(store) + PrivateCoordinatorTable.install(store) store.participants.register( alpha.incarnation.lookup, "alpha", "alpha", committed=True ) diff --git a/tests/test_busy_history_loading.py b/tests/test_busy_history_loading.py index 7d96f548e4..e3ea7e4579 100644 --- a/tests/test_busy_history_loading.py +++ b/tests/test_busy_history_loading.py @@ -1,5 +1,6 @@ """Source contention retains a viewer read, never manufactures an input failure.""" +from agent_comms.private_runtime_schema import PrivateCoordinatorTable import asyncio import json import subprocess @@ -28,7 +29,7 @@ def saved_history(tmp_path): ))) comms.registry.declare(comms.registry.new_thread('saved', frozenset(), str(tmp_path), session_file=str(session))) with Coordination(str(comms.root / "coordination.sqlite3")) as store: - store.install_private_runtime() + PrivateCoordinatorTable.install(store) return comms, session diff --git a/tests/test_certified_bootstrap.py b/tests/test_certified_bootstrap.py index 4dfcdbe067..c78d3c384b 100644 --- a/tests/test_certified_bootstrap.py +++ b/tests/test_certified_bootstrap.py @@ -1,5 +1,6 @@ """Fresh roots refuse unproven source coverage and unconfigured native owners.""" +from agent_comms.private_runtime_schema import PrivateCoordinatorTable import os import pytest @@ -7,7 +8,6 @@ from agent_comms.acp import CommsAgent from agent_comms.comms import Comms from agent_comms.channels import AnyChannel -from agent_comms.coordinated_runtime_schema import install_native_runtime_schema from agent_comms.coordination_cohort import accept_delivery_cohort from agent_comms.coordination_errors import IdentityConflict, PublicationActivationBlocked from agent_comms.proven_source_coverage import SourceCoverage @@ -19,7 +19,7 @@ def test_fresh_root_claims_source_gaps_and_empty_cursor(tmp_path, *, source_reads): comms, store, root_id, lookups = _root(tmp_path) with store: - install_native_runtime_schema(store) + PrivateCoordinatorTable.install(store) assert comms.bus.bus.revision() == 0 empty = SourceCoverage( comms.bus, store, wire_root_id=root_id, recipient_lookup=lookups["bob"] @@ -41,7 +41,7 @@ def test_fresh_root_claims_source_gaps_and_empty_cursor(tmp_path, *, source_read def test_truncated_coverage_cannot_hide_an_unproven_source(tmp_path, *, source_reads): comms, store, root_id, lookups = _root(tmp_path) with store: - install_native_runtime_schema(store) + PrivateCoordinatorTable.install(store) first = comms.messaging.send_user_message("bob", "first original", worktree=str(tmp_path)) second = comms.messaging.send_user_message("bob", "second original", worktree=str(tmp_path)) reader = SourceCoverage( diff --git a/tests/test_claim_admission_verifier.py b/tests/test_claim_admission_verifier.py index 2fd3ddc81c..7510471afd 100644 --- a/tests/test_claim_admission_verifier.py +++ b/tests/test_claim_admission_verifier.py @@ -2,6 +2,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from bus_record_fixture import prepare_native_comms from conftest import publish_initial @@ -23,7 +24,6 @@ verify_selected_wake, write_selected_claimed_file, ) -from agent_comms.cohort_schema import install_private_cohort_schema from agent_comms.comms import Comms from agent_comms.coordination_cohort import accept_delivery_cohort, sealed_cohort_assignments from agent_comms.coordination_errors import IdentityConflict @@ -61,7 +61,7 @@ def test_selected_wake_verifier_refuses_no_wake_and_stale_authority( message = publish_initial(comms, "sender", "#team", "@Alice investigate") initial = comms.bus.bus.delivery(message.seq) with Coordination(str(root / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) for recipient in initial.audience.recipients: store.participants.register( recipient.recipient_lookup, diff --git a/tests/test_claim_bus_read.py b/tests/test_claim_bus_read.py index 66bdc25251..0a44fbc6e5 100644 --- a/tests/test_claim_bus_read.py +++ b/tests/test_claim_bus_read.py @@ -8,6 +8,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from agent_comms.context_store import UNREAD_CONTEXT from conftest import publish_initial @@ -27,7 +28,6 @@ publish_selected_resource_claim, write_selected_claimed_file, ) -from agent_comms.cohort_schema import install_private_cohort_schema from agent_comms.comms import Comms from agent_comms.coordination_cohort import accept_delivery_cohort, sealed_cohort_assignments from agent_comms.coordination_errors import IdentityConflict @@ -89,7 +89,7 @@ def selected_turn(private_comms, private_directory): record_context(comms, "Alice", 2) initial = comms.bus.bus.delivery(message.seq) with Coordination(str(root / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) for recipient in initial.audience.recipients: store.participants.register( recipient.recipient_lookup, recipient.canonical_thread, diff --git a/tests/test_cohort_schema.py b/tests/test_cohort_schema.py index 1697646d94..fa1dfc589c 100644 --- a/tests/test_cohort_schema.py +++ b/tests/test_cohort_schema.py @@ -2,6 +2,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable import os import sqlite3 from concurrent.futures import ThreadPoolExecutor @@ -10,7 +11,7 @@ import pytest -from agent_comms.cohort_schema import CohortTable, install_private_cohort_schema +from agent_comms.cohort_schema import CohortTable from agent_comms.coordination_errors import SchemaVersionError from agent_comms.coordinator import Coordination @@ -135,13 +136,13 @@ def test_selected_claims_follow_ordered_n_subsequence_not_only_set_membership( tmp_path: Path, ) -> None: with Coordination(str(tmp_path / "valid.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) with store.session.writing() as db: _seal_two_selected_with_observer(db, reverse_claims=False) sealed = store.session._connection.execute("SELECT sealed FROM claim_batch_receipts").fetchone()[0] assert sealed == 1 with Coordination(str(tmp_path / "reversed.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) with ( pytest.raises(sqlite3.IntegrityError), store.session.writing() as db, @@ -156,15 +157,15 @@ def test_schema_is_opt_in_declared_and_idempotent_on_reopen(tmp_path: Path) -> N with Coordination(str(path)) as store: assert "claim_batch_receipts" not in _tables(store) assert "cohort_delivery_receipts" not in _tables(store) - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) before = set(_tables(store)) - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) assert _tables(store) == before if os.name == "posix": assert path.stat().st_mode & 0o077 == 0 # Windows' st_mode does not attest NTFS ACL protection. with Coordination(str(path)) as reopened: - install_private_cohort_schema(reopened) + PrivateCoordinatorTable.install(reopened) assert _tables(reopened) == before assert reopened.session._connection.execute("PRAGMA journal_mode").fetchone()[0] == "delete" assert reopened.session._connection.execute("PRAGMA foreign_keys").fetchone()[0] == 1 @@ -174,7 +175,7 @@ def test_schema_is_opt_in_declared_and_idempotent_on_reopen(tmp_path: Path) -> N def test_zero_member_zero_claim_cohort_is_durable_without_fabricated_claim(tmp_path: Path) -> None: path = tmp_path / "coordination.sqlite3" with Coordination(str(path)) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) with store.session.writing() as db: _receipt(db, n=0, k=0) _seal(db) @@ -183,7 +184,7 @@ def test_zero_member_zero_claim_cohort_is_durable_without_fabricated_claim(tmp_p ) assert store.session._connection.execute("SELECT COUNT(*) FROM wake_claims").fetchone()[0] == 0 with Coordination(str(path)) as reopened: - install_private_cohort_schema(reopened) + PrivateCoordinatorTable.install(reopened) assert ( reopened.session._connection.execute("SELECT COUNT(*) FROM claim_batch_receipts").fetchone()[0] == 1 @@ -192,7 +193,7 @@ def test_zero_member_zero_claim_cohort_is_durable_without_fabricated_claim(tmp_p def test_seal_rejects_observer_with_preexisting_legacy_singleton_claim(tmp_path: Path) -> None: with Coordination(str(tmp_path / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) with store.session.writing() as db: _assignment(db) db.execute( @@ -233,7 +234,7 @@ def test_selected_and_observer_receipts_are_distinct_immutable_rows(tmp_path: Pa from agent_comms.field_codec import FieldCodec with Coordination(str(tmp_path / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) with store.session.writing() as db: _assignment(db) _receipt(db, n=2, k=1) @@ -277,7 +278,7 @@ def test_selected_and_observer_receipts_are_distinct_immutable_rows(tmp_path: Pa def test_seal_rejects_missing_rows_and_rolls_back_entire_cohort(tmp_path: Path) -> None: with Coordination(str(tmp_path / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) with pytest.raises(sqlite3.IntegrityError): # noqa: SIM117 with store.session.writing() as db: _assignment(db) @@ -328,7 +329,7 @@ def test_seal_rejects_missing_rows_and_rolls_back_entire_cohort(tmp_path: Path) def test_foreign_claim_lookup_and_kind_shape_fail_before_seal(tmp_path: Path) -> None: with Coordination(str(tmp_path / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) with store.session.writing() as db: _assignment(db) _receipt(db, n=2, k=1) @@ -404,7 +405,7 @@ def test_seal_cannot_adopt_claim_with_conflicting_immutable_acceptance( tmp_path: Path, field: str, override: str | int ) -> None: with Coordination(str(tmp_path / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) with ( pytest.raises(sqlite3.IntegrityError), store.session.writing() as db, @@ -425,7 +426,7 @@ def test_seal_cannot_adopt_claim_with_conflicting_immutable_acceptance( def test_one_claim_cannot_be_sealed_by_two_wire_roots(tmp_path: Path) -> None: with Coordination(str(tmp_path / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) with store.session.writing() as db: _assignment(db) _receipt(db, n=1, k=1) @@ -455,11 +456,11 @@ def test_reopen_rejects_partial_or_drifted_schema(tmp_path: Path, problem: str) if problem == "partial": store.session._connection.execute("CREATE TABLE claim_batch_members (x INTEGER)") else: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) store.session._connection.execute("DROP INDEX cohort_delivery_receipts_recipient_lookup_wire_seq_idx") before = set(_tables(store)) with pytest.raises(SchemaVersionError): - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) assert _tables(store) == before assert not store.session._connection.in_transaction @@ -473,7 +474,7 @@ def test_two_process_like_connections_serialize_explicit_migration(tmp_path: Pat def migrate() -> int: with Coordination(path) as store: barrier.wait(timeout=5) - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) return CohortTable.schema_differs(store.session._connection) with ThreadPoolExecutor(max_workers=2) as pool: diff --git a/tests/test_coordinated_runtime.py b/tests/test_coordinated_runtime.py index cf8d5ecbc9..d62c5bd78a 100644 --- a/tests/test_coordinated_runtime.py +++ b/tests/test_coordinated_runtime.py @@ -2,6 +2,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable import json import os from dataclasses import replace @@ -13,7 +14,6 @@ from agent_comms.comms import Comms from agent_comms.context_segments.wake import SelectedWakeSegment from agent_comms.coordinated_runtime import SelectedExecution, SessionSetupFailed -from agent_comms.coordinated_runtime_schema import install_native_runtime_schema from agent_comms.native_runtime_input import NativeRuntimeInput, NativeRuntimeTable from agent_comms.coordination_cohort import accept_delivery_cohort, sealed_cohort_assignments from agent_comms.coordination_errors import ( @@ -161,8 +161,8 @@ def test_native_runtime_schema_explicit_install_and_drift_fail_closed(tmp_path: ).fetchone() is None ) - install_native_runtime_schema(store) - install_native_runtime_schema(store) + PrivateCoordinatorTable.install(store) + PrivateCoordinatorTable.install(store) NativeRuntimeTable.require_installed(store.session._connection) trigger = next(iter(NativeRuntimeInput.triggers())) store.session._connection.execute(f'DROP TRIGGER "{trigger}"') diff --git a/tests/test_coordination_cohort.py b/tests/test_coordination_cohort.py index 5c14e44487..cdab6d9f93 100644 --- a/tests/test_coordination_cohort.py +++ b/tests/test_coordination_cohort.py @@ -2,6 +2,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from bus_record_fixture import prepare_native_comms from agent_comms.bus import Bus @@ -24,9 +25,7 @@ DeliveredAssignment, FullPendingAssignment, ) -from agent_comms.cohort_schema import ( - ClaimBatchAcknowledgements, CohortDeliveries, install_private_cohort_schema, -) +from agent_comms.cohort_schema import ClaimBatchAcknowledgements, CohortDeliveries from agent_comms.messages import Message from agent_comms.private_registry_guard import RegistryPrivacy from agent_comms.comms import Comms @@ -74,7 +73,7 @@ def _root(tmp_path: Path) -> tuple[Comms, Coordination, str, dict[str, str]]: for name, (created_at, _tags) in people.items() } coordinator = Coordination(str(root / "coordinator.sqlite"), clock_ms=lambda: 4723) - install_private_cohort_schema(coordinator) + PrivateCoordinatorTable.install(coordinator) for name in ("Alice", "Bob", "Charlie"): coordinator.participants.register(lookups[name], name, name, committed=True) root_id = comms.registry.prepare_native_root().namespace diff --git a/tests/test_coordination_response.py b/tests/test_coordination_response.py index 3f89ef2ed2..e9fd6520c0 100644 --- a/tests/test_coordination_response.py +++ b/tests/test_coordination_response.py @@ -2,6 +2,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from bus_record_fixture import prepare_native_comms from agent_comms.lifecycle_capability import Active, Complete @@ -16,7 +17,6 @@ from agent_comms.attempt_start import AttemptStart from agent_comms.attempt_states import ModelRunningAttempt, PromptAcceptedAttempt, SettlingAttempt from agent_comms.child_process import ProcessIdentity -from agent_comms.cohort_schema import install_private_cohort_schema from agent_comms.comms import Comms from agent_comms.coordination_cohort import accept_delivery_cohort from agent_comms.coordination_errors import ( @@ -24,13 +24,7 @@ PublicationUncertain, StaleRevision, ) -from agent_comms.coordination_response import ( - LiveResponseOwner, - install_private_response_schema, - prepare_fenced_response, - publish_fenced_response, - resolve_existing_response, -) +from agent_comms.coordination_response import LiveResponseOwner, prepare_fenced_response, publish_fenced_response, resolve_existing_response from agent_comms.coordination_results import AlreadyApplied from agent_comms.coordination_tables.executions import ExecutionOrigin from agent_comms.coordinator import Coordination @@ -82,8 +76,7 @@ def _ready(tmp_path: Path, *, direct: bool = False, extra_targets: tuple[str, .. recipient = original_record.audience.recipients[0] bus = responder(comms.root, comms.registry) store = Coordination(str(comms.root / "coordination.sqlite3"), clock_ms=lambda: 1_000) - install_private_cohort_schema(store) - install_private_response_schema(store) + PrivateCoordinatorTable.install(store) store.participants.register( recipient.recipient_lookup, recipient.canonical_thread, diff --git a/tests/test_declarations.py b/tests/test_declarations.py index 8501eed36b..8b28d476a5 100644 --- a/tests/test_declarations.py +++ b/tests/test_declarations.py @@ -1,4 +1,5 @@ +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from agent_comms.bus import Bus from bus_record_fixture import prepare_native_comms @@ -397,7 +398,7 @@ def test_private_guard_rejects_disk_downgrade_even_with_cached_revision( prepare_native_comms(comms) # A native coordinator is retained evidence of private mode even if its guard vanishes. with Coordination(str(CoordinationStore.path_in(root))) as store: - CohortTable.install(store) + PrivateCoordinatorTable.install(store) cold = Registration(root / "registry.json") assert cold.require("a").name == "a" guard = root / ".registry-owner-guard" diff --git a/tests/test_install_announcements.py b/tests/test_install_announcements.py index 5bbfd8cb23..bb2ca9c9ae 100644 --- a/tests/test_install_announcements.py +++ b/tests/test_install_announcements.py @@ -52,7 +52,8 @@ def test_failed_rehearsal_does_not_publish_to_the_source_bus(private_comms, tmp_ with pytest.raises(UpgradeRefused, match="local changes"): upgrade( source.checkout, private_comms.root, target, UnnamedInstaller(), True, False, - tmp_path / "spill", tmp_path / "scratch", lambda: None, + tmp_path / "spill", tmp_path / "scratch", + lambda: pytest.fail("A refused rehearsal cannot install harnesses"), InstallReport(private_comms.root), ) diff --git a/tests/test_native_evidence_custody.py b/tests/test_native_evidence_custody.py index 43651eb6f8..283b1fccbd 100644 --- a/tests/test_native_evidence_custody.py +++ b/tests/test_native_evidence_custody.py @@ -1,5 +1,6 @@ """Original-byte ownership survives appends and refuses altered proof/source.""" +from agent_comms.private_runtime_schema import PrivateCoordinatorTable import asyncio import json import os @@ -167,7 +168,7 @@ async def test_joined_worker_retains_original_evidence_and_owns_sqlite_until_can before = session.read_bytes(), files.input_proof.read_bytes() path = tmp_path / 'coordination.sqlite3' with Coordination(str(path), clock_ms=lambda: 1234) as original: - original.install_private_runtime() + PrivateCoordinatorTable.install(original) original.participants.register('original', 'owner', 'owner', committed=True) original_connection = original.session._connection entered, release = threading.Event(), threading.Event() @@ -178,7 +179,7 @@ def observe(store): assert store.session.now() == original.session.now() worker_connections.append(store.session._connection) with store.session.read(): - assert store.participants.get('original').owner_thread == 'owner' + assert store.participants.get('original').owner_identity.owner_thread == 'owner' NativeContextProof.read_evidence(files, INPUT_ID, evidence=evidence, session_entry_start=input_record.start) entered.set() @@ -203,7 +204,7 @@ def observe(store): with pytest.raises(sqlite3.ProgrammingError): _ = worker_connections[0].in_transaction with original.session.read(): - assert original.participants.get('original').owner_thread == 'owner' + assert original.participants.get('original').owner_identity.owner_thread == 'owner' finally: release.set() if not pending.done(): diff --git a/tests/test_optional_awareness_projection.py b/tests/test_optional_awareness_projection.py index 69154049e5..1f1f0cc287 100644 --- a/tests/test_optional_awareness_projection.py +++ b/tests/test_optional_awareness_projection.py @@ -2,6 +2,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from bus_record_fixture import prepare_native_comms from agent_comms.lifecycle_capability import Executing @@ -16,11 +17,7 @@ from agent_comms.thread_identity import RecordedThreadIncarnation from agent_comms.bus_publication import Delivered from agent_comms.child_process import ProcessIdentity -from agent_comms.cohort_schema import ( - AwarenessClaimGenerations, - ClaimBatchAcknowledgements, - install_private_cohort_schema, -) +from agent_comms.cohort_schema import AwarenessClaimGenerations, ClaimBatchAcknowledgements from agent_comms.comms import Comms from agent_comms.coordination_cohort import accept_delivery_cohort from agent_comms.coordination_errors import IdentityConflict @@ -56,7 +53,7 @@ def _root( ) root_id = comms.registry.prepare_native_root().namespace store = Coordination(str(root / "coordination.sqlite3")) - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) return comms, store, WakeCandidateIndex(comms.bus.bus), root_id diff --git a/tests/test_private_human_ingress.py b/tests/test_private_human_ingress.py index 09bfa99f95..1dfb1bf872 100644 --- a/tests/test_private_human_ingress.py +++ b/tests/test_private_human_ingress.py @@ -2,6 +2,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from bus_record_fixture import prepare_native_comms from agent_comms.channel_targets import CommandTarget @@ -15,7 +16,6 @@ from agent_comms.thread_identity import RecordedThreadIncarnation from agent_comms.bus_publication import HumanOrigin -from agent_comms.cohort_schema import install_private_cohort_schema from agent_comms.comms import Comms, wire from agent_comms.coordination_cohort import accept_delivery_cohort from agent_comms.coordination_results import Applied @@ -47,7 +47,7 @@ def _root(tmp_path: Path) -> tuple[Comms, Coordination, str, dict[str, str]]: comms.registry.declare(comms.registry.new_thread(name, frozenset({'team'}), str(root), created_at=created_at)) lookups[name] = RecordedThreadIncarnation("fixture", created_at).lookup store = Coordination(str(CoordinationStore.path_in(root)), clock_ms=lambda: 9999) - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) for name in lookups: store.participants.register(lookups[name], name, name, committed=True) return comms, store, comms.registry.prepare_native_root().namespace, lookups diff --git a/tests/test_reconcile_stores.py b/tests/test_reconcile_stores.py index 7f797243a5..026ee08244 100644 --- a/tests/test_reconcile_stores.py +++ b/tests/test_reconcile_stores.py @@ -20,6 +20,13 @@ ReconcileStoresCliCommand, ) from agent_comms.activity import ActivityStore +from agent_comms.coordination_database import CoordinationStore +from agent_comms.coordination_session import CoordinationSession +from agent_comms.coordinator import Coordination +from agent_comms.native_runtime_input import NativeRuntimeInput +from agent_comms.private_runtime_schema import PrivateCoordinatorSchema +from agent_comms.triage_native_sources import TriageNativeSources +from agent_comms.bus import Bus from agent_comms.maintenance_barrier import HOLDER_NONCE from agent_comms.installer.backup import bus_files from agent_comms.typed_table import ReconciledFile @@ -107,3 +114,47 @@ def test_read_back_under_the_hold_writes_nothing_on_a_drifted_bus(private_comms) completed = reconcile(root, nonce, ReadBackCliCommand) assert completed.returncode == 0, completed.stdout + completed.stderr assert bus_files(root) == before + + +def test_private_carry_plans_and_restores_missing_native_tables(private_comms): + root = private_comms.root + database = CoordinationStore.path_in(root) + with Coordination(database) as store: + store.participants.register("retained", "Retained owner", "retained-thread", committed=True) + retained = store.participants.get("retained") + store.session._connection.execute(f"DROP TABLE {TriageNativeSources.declared_name}") + + with maintenance_held(root) as nonce: + before = bus_files(root) + planned = reconcile(root, nonce, PlannedReconcileCliCommand) + assert planned.returncode == 0, planned.stdout + planned.stderr + assert json.loads(planned.stdout)["rewritten"] == [database.name] + assert bus_files(root) == before + carried = reconcile(root, nonce) + assert carried.returncode == 0, carried.stdout + carried.stderr + assert json.loads(carried.stdout)["rewritten"] == [database.name] + assert json.loads(reconcile(root, nonce, PlannedReconcileCliCommand).stdout)["rewritten"] == [] + + with Coordination(database) as store: + assert store.participants.get("retained") == retained + PrivateCoordinatorSchema.require_installed(store.session._connection) + + +def test_public_reconcile_does_not_install_native_state(tmp_path): + root = tmp_path / "public" + Bus.install(Bus.path_in(root)) + database = CoordinationStore.path_in(root) + with CoordinationSession(database): + pass + before = bus_files(root) + planned = reconcile(root, command=PlannedReconcileCliCommand) + assert planned.returncode == 0, planned.stdout + planned.stderr + assert json.loads(planned.stdout)["rewritten"] == [] + carried = reconcile(root) + assert carried.returncode == 0, carried.stdout + carried.stderr + assert json.loads(carried.stdout)["rewritten"] == [] + assert bus_files(root) == before + with closing(sqlite3.connect(database)) as db: + assert not db.execute( + "SELECT name FROM sqlite_master WHERE name=?", (NativeRuntimeInput.declared_name,) + ).fetchone() diff --git a/tests/test_roster_restoration.py b/tests/test_roster_restoration.py index f27d438bf3..96c179095a 100644 --- a/tests/test_roster_restoration.py +++ b/tests/test_roster_restoration.py @@ -1,5 +1,6 @@ """Restoration changes visibility without replay or transferring live ownership.""" +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from bus_record_fixture import prepare_native_comms from conftest import installed_comms @@ -123,7 +124,6 @@ def test_catalog_restore_preserves_current_preferences_and_historical_channels(t def test_private_restoration_allows_new_cohort_without_starting_old_subscribers( tmp_path, repair_existing ): - from agent_comms.cohort_schema import install_private_cohort_schema from agent_comms.coordination_cohort import accept_delivery_cohort from agent_comms.coordinator import Coordination @@ -135,7 +135,7 @@ def test_private_restoration_allows_new_cohort_without_starting_old_subscribers( root_id = current.registry.prepare_native_root().namespace source = old.registry.snapshot() with Coordination(str(current.root / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) live_before = store.participants.register( RecordedThreadIncarnation("fixture", live.created_at).lookup, live.name, live.name, committed=True ).value diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 2e6c9160eb..5710fdf9aa 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -1,5 +1,6 @@ """Single-owner attachment and idle wakeup through real processes and sockets.""" +from agent_comms.private_runtime_schema import PrivateCoordinatorTable import asyncio import json import os @@ -133,7 +134,7 @@ async def session_update(self, session_id, update): ))) thread = comms.registry.require(original) with Coordination(str(comms.root / "coordination.sqlite3")) as store: - store.install_private_runtime() + PrivateCoordinatorTable.install(store) store.participants.register(thread.incarnation.lookup, original, original, committed=True) comms.threads.attach_session(thread, str(journal), process=thread.process_identity) comms.threads.rename_managed_thread(original, "renamed", owner_pid=os.getpid()) diff --git a/tests/test_selected_answer_forms.py b/tests/test_selected_answer_forms.py index 2dd8c8287a..b276ac0400 100644 --- a/tests/test_selected_answer_forms.py +++ b/tests/test_selected_answer_forms.py @@ -8,6 +8,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable from bus_record_fixture import prepare_native_comms from conftest import publish_initial @@ -22,7 +23,6 @@ from agent_comms.thread_identity import RecordedThreadIncarnation from agent_comms.child_process import ProcessIdentity -from agent_comms.cohort_schema import install_private_cohort_schema from agent_comms.comms import Comms from agent_comms.coordination_cohort import accept_delivery_cohort, sealed_cohort_assignments from agent_comms.coordinator import Coordination @@ -48,7 +48,7 @@ def alice_batch(*sent: tuple[str, str]): root_id = comms.registry.prepare_native_root().namespace messages = [publish_initial(comms, "sender", target, text) for target, text in sent] with Coordination(str(root / "coordination.sqlite3")) as store: - install_private_cohort_schema(store) + PrivateCoordinatorTable.install(store) deliveries = [comms.bus.bus.delivery(m.seq) for m in messages] for recipient in deliveries[0].audience.recipients: store.participants.register( diff --git a/tests/test_selected_tool_broker.py b/tests/test_selected_tool_broker.py index 2c7df9768b..2268ab6478 100644 --- a/tests/test_selected_tool_broker.py +++ b/tests/test_selected_tool_broker.py @@ -2,6 +2,7 @@ from __future__ import annotations +from agent_comms.private_runtime_schema import PrivateCoordinatorTable import asyncio import json import os @@ -14,7 +15,6 @@ import agent_comms.selected_tool_broker as broker from agent_comms.channel_coding_tools import CodingToolSocket from agent_comms.comms import Comms -from agent_comms.coordinated_runtime_schema import install_native_runtime_schema from agent_comms.coordinator import Coordination from agent_comms.envelope_claim_transitions import WakeClaim from agent_comms.native_tool_call import ( @@ -108,7 +108,7 @@ def test_owner_mode_denies_unreserved_input_before_consuming_ledger(tmp_path: Pa attempt_ordinal=1, ) with Coordination(str(root / "coordination.sqlite3")) as store: - install_native_runtime_schema(store) + PrivateCoordinatorTable.install(store) input_id = secrets.token_hex(16) mode = broker.selected_tool_mode_for_owner( comms, store, claim_scope, "owner", session_dir, input_id diff --git a/tests/test_tool_refusal_reasons.py b/tests/test_tool_refusal_reasons.py index b19640c377..b85bd7a773 100644 --- a/tests/test_tool_refusal_reasons.py +++ b/tests/test_tool_refusal_reasons.py @@ -1,5 +1,6 @@ """A refused coding tool call tells the agent the real reason.""" +from agent_comms.private_runtime_schema import PrivateCoordinatorTable import secrets import sqlite3 from pathlib import Path @@ -34,7 +35,7 @@ def test_busy_coordination_database_is_reported_as_busy(tmp_path): (tmp_path / "wire").mkdir(mode=0o700) path = tmp_path / "wire" / "coordination.sqlite3" with Coordination(str(path)) as store: - store.install_private_runtime() + PrivateCoordinatorTable.install(store) writer = sqlite3.connect(path, isolation_level=None) try: writer.execute("BEGIN EXCLUSIVE") diff --git a/tests/test_transcript_partial_tail.py b/tests/test_transcript_partial_tail.py index ae36ea3e79..0ddfab7b81 100644 --- a/tests/test_transcript_partial_tail.py +++ b/tests/test_transcript_partial_tail.py @@ -1,5 +1,6 @@ """A line still being appended is not history yet, so a newer read never stalls on it.""" +from agent_comms.private_runtime_schema import PrivateCoordinatorTable import json from agent_comms.thread_text import SavedSessionFile @@ -23,7 +24,7 @@ def test_newer_read_moves_past_a_line_still_being_appended(tmp_path): "timestamp": "2026-10-20T00:00:00.000Z"}) + line(answer("first", "First answer"))) comms.registry.declare(comms.registry.new_thread("saved", frozenset(), str(tmp_path), session_file=SavedSessionFile(str(session)))) with Coordination(str(comms.root / "coordination.sqlite3")) as store: - store.install_private_runtime() + PrivateCoordinatorTable.install(store) complete = session.stat().st_size pending = line(answer("second", "Second answer " + "x" * 200_000)) with session.open("a") as stream: From 55448907dd8e0f2a404879a06b28bc180b570d61 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 21:03:32 -0400 Subject: [PATCH 3/4] Omit pre-rename captures from optional current awareness --- src/agent_comms/cohort_schema.py | 8 +++----- tests/test_optional_awareness_projection.py | 13 +++++++++++++ 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/src/agent_comms/cohort_schema.py b/src/agent_comms/cohort_schema.py index 0d69447ebc..e4954330f9 100644 --- a/src/agent_comms/cohort_schema.py +++ b/src/agent_comms/cohort_schema.py @@ -315,14 +315,12 @@ class AwarenessClaimGenerations( owner_generation: Annotated[int, Column(check="owner_generation > 0")] def current(self, expected: OwnerGenerations) -> bool: - """A historical capture is omitted; a changed current owner refuses.""" + """Old generations and pre-rename names are historical, never current context.""" from .wake_candidate_index import ProjectionUnavailableError - if self.owner_generation < expected.generation: - return False - if self.owner_identity != expected: + if self.owner_lookup != expected.owner_lookup or self.owner_generation > expected.generation: raise ProjectionUnavailableError("captured participant owner changed") - return True + return self.owner_identity == expected def require_selection(self, assignment: WakeAssignment, acknowledgement: ClaimBatchAcknowledgements) -> None: """Verify this immutable acceptance capture against its original claim.""" diff --git a/tests/test_optional_awareness_projection.py b/tests/test_optional_awareness_projection.py index 1f1f0cc287..d159ca7171 100644 --- a/tests/test_optional_awareness_projection.py +++ b/tests/test_optional_awareness_projection.py @@ -32,6 +32,7 @@ from agent_comms.threads import Thread from agent_comms.wake_candidate_index import ( WakeCandidateIndex, + ProjectionUnavailableError, ) from agent_comms.thread_model import DeclaredModel @@ -317,6 +318,18 @@ def test_normal_rename_does_not_inject_old_selected_claim_into_new_owner( selected = result.context()["selected"] assert [row["claim_id"] for row in selected] == [current_claim.assignment_id] assert old_claim.assignment_id not in result.render(32768) + captured = AwarenessClaimGenerations.one( + store.session._connection, claim_id=old_claim.assignment_id, + ) + current_owner = store.participants.get(current_claim.recipient_lookup).owner_identity + # A supported rename preserves generation, but it does not make unrelated + # lookups or a capture from a future generation historical evidence. + for invalid in ( + replace(captured, owner_lookup="unrelated-owner"), + replace(captured, owner_generation=current_owner.generation + 1), + ): + with pytest.raises(ProjectionUnavailableError): + invalid.current(current_owner) finally: store.close() From 504c7d530e6c44571cd4fd992f30f7d270fe5b2e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 21:08:21 -0400 Subject: [PATCH 4/4] Repair standalone native digest reads and qualify native consumers --- src/agent_comms/native_pi.py | 4 +-- tests/test_native_failure_recovery.py | 12 +++++-- tests/test_selected_execution_native.py | 48 +++++++++++++++++-------- 3 files changed, 45 insertions(+), 19 deletions(-) diff --git a/src/agent_comms/native_pi.py b/src/agent_comms/native_pi.py index 810a86cbae..f3de4c8b32 100644 --- a/src/agent_comms/native_pi.py +++ b/src/agent_comms/native_pi.py @@ -891,9 +891,9 @@ def read_tracked_input_commitment( """ Id128Text.decode(input_id) session_file = Path(session_file).absolute() - from .native_entries import NativeInputEvidenceRead + from .native_entries import NativeEvidenceRead - with NativeInputEvidenceRead.borrow(session_file, evidence) as evidence: + with NativeEvidenceRead.borrow(session_file, evidence) as evidence: evidence.observe() try: return CommittedInput(evidence.tracked(input_id).message.input_digest.require()) diff --git a/tests/test_native_failure_recovery.py b/tests/test_native_failure_recovery.py index 44d7f89de6..f24abd6dcf 100644 --- a/tests/test_native_failure_recovery.py +++ b/tests/test_native_failure_recovery.py @@ -113,6 +113,7 @@ def record(cls, claim, session, participant): NativeRuntimeInput.one(db, input_id=claim.input_id).commit_context(db, context) append_input_entry(identity.path, { "type": "message", "id": "error-terminal", "parentId": entry_id, + "timestamp": "2026-10-01T00:00:00.001Z", "message": {"role": "assistant", "stopReason": "error", "errorMessage": "Codex error: The usage limit has been reached", "content": []}, @@ -226,12 +227,19 @@ def alter_rows(self, rows): class AdditionalInputDamage(EvidenceDamage): def alter_rows(self, rows): - rows.append({"type": "message", "message": {"role": "user", "content": []}}) + rows.append({ + "type": "message", "id": "additional-input", "parentId": rows[-1]["id"], + "timestamp": "2026-10-01T00:00:00.002Z", + "message": {"role": "user", "content": []}, + }) class ToolDamage(EvidenceDamage): def alter_rows(self, rows): - rows[-1]["message"]["content"] = [{"type": "toolCall", "name": "bash"}] + rows[-1]["message"]["content"] = [{ + "type": "toolCall", "id": "unexpected-tool", "name": "bash", + "arguments": {"command": "true"}, + }] class WrongEpochDamage(EvidenceDamage): diff --git a/tests/test_selected_execution_native.py b/tests/test_selected_execution_native.py index e922c19028..5fbf469ae4 100644 --- a/tests/test_selected_execution_native.py +++ b/tests/test_selected_execution_native.py @@ -9,6 +9,7 @@ import os import shlex import threading +import traceback from dataclasses import replace from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path @@ -19,18 +20,19 @@ from agent_comms.coordinated_runtime import SelectedExecution from agent_comms.coordination_errors import IdentityConflict from agent_comms.coordinator import Coordination -from agent_comms.field_codec import FieldCodec +from agent_comms.agent_activity import AgentActivity from agent_comms.historical_native_inputs import read_historical_native_inputs from agent_comms.native_input_record import TriageNativeExecution, FullNativeExecution from agent_comms.native_pi import CommittedInput, read_tracked_input_commitment from agent_comms.native_package import VerifiedNativePackage -from agent_comms.messages import Message, InfoKind +from agent_comms.native_tools import NativeToolDeclaration from agent_comms.cursor_owner import CursorOwner from agent_comms.selected_tool_broker import SelectedToolIntent from test_coordinated_runtime import _root from test_coordinated_runtime import tmp_path as private_root_fixture from agent_comms.thread_model import DeclaredModel +from agent_comms.turn_phase import ModelWaitPhase, ToolRunningPhase tmp_path = private_root_fixture @@ -50,6 +52,19 @@ async def test_native_full_four_tools_publish_and_release(tmp_path, monkeypatch, requests = [] failures = [] phases = [] + native_phases = [] + model_wait_after_tools = threading.Event() + observe_native_phase = AgentActivity.observe_native_phase + + def record_native_phase(agents, lease, phase): + states = observe_native_phase(agents, lease, phase) + if agents.registry.store.path.parent == root and states: + native_phases.append(type(lease.record.read().require_active().phase)) + if native_phases[-2:] == [ToolRunningPhase, ModelWaitPhase]: + model_wait_after_tools.set() + return states + + monkeypatch.setattr(AgentActivity, "observe_native_phase", record_native_phase) calls = [ ("read", {"path": "input.txt"}), ("edit", {"path": "input.txt", "edits": [{"oldText": "BEFORE", "newText": "AFTER"}]}), @@ -74,7 +89,7 @@ async def test_native_full_four_tools_publish_and_release(tmp_path, monkeypatch, 'from pathlib import Path; from agent_comms.comms import Comms; from agent_comms.publisher import Publisher; ' 'from agent_comms.turn_phase import ToolRunningPhase; ' f'c=Comms(Path({str(root)!r}), Publisher); ' - 'assert isinstance(c.registry.require("beta").active_turn.phase, ToolRunningPhase); ' + 'assert isinstance(c.registry.require("beta").turn_record.read().require_active().phase, ToolRunningPhase); ' 'print("CANONICAL_TOOL_RUNNING")' ) python = shlex.quote(str(Path(entrypoint).with_name('python'))) @@ -90,13 +105,13 @@ def do_POST(self): try: request = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) requests.append(request) - phase = comms.registry.require('beta').active_turn.phase + phase = comms.registry.require('beta').turn_record.read().require_active().phase phases.append(phase.declared_name) assert request["model"] == "fixture" assert self.headers["Authorization"] == "Bearer offline-only-fixture" assert len(requests) <= 2 + triage if triage and len(requests) == 1: - assert "bounded triage" in json.dumps(request) + assert "Output ONLY a JSON object with one key" in json.dumps(request) assert not request.get("tools") delta = {"role": "assistant", "content": '{"decision":"FULL"}'} reason = "stop" @@ -106,9 +121,11 @@ def do_POST(self): assert "Selected source decisions through " in context assert "Optional non-authoritative awareness" in context assert initial.message.message_id in context - assert {t["function"]["name"] for t in request["tools"]} == { - name for name, _ in calls - } + declared = ({name for name, _ in calls} if selected_write else { + tool.declared_name + for tool in NativeToolDeclaration.members_with(NativeToolDeclaration) + }) + assert {t["function"]["name"] for t in request["tools"]} == declared delta = { "role": "assistant", "tool_calls": [ @@ -131,11 +148,12 @@ def do_POST(self): assert not any("Error:" in str(t["content"]) for t in tools) if os.environ.get('AC_HEADLESS_ENTRYPOINT'): assert 'CANONICAL_TOOL_RUNNING' in str(tools[-1]['content']) - assert phase.declared_name == 'model_wait' + # Pi's next HTTP request can outrun Core's observation of + # the tool terminal. Hold this response until the real + # phase owner commits the transition back to model wait. + assert model_wait_after_tools.wait(5), native_phases assert 'participant' not in comms.registry.snapshot().threads - answer = Message("beta", initial.message.exact_reply_target, - "CODING_TOOLS_OK", InfoKind, timestamp=0) - delta = {"role": "assistant", "content": json.dumps(FieldCodec.encode((answer,)))} + delta = {"role": "assistant", "content": "CODING_TOOLS_OK"} reason = "stop" chunk = { "id": "selected-offline", @@ -158,8 +176,8 @@ def do_POST(self): self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) - except Exception as error: - failures.append(str(error)) + except Exception: + failures.append(traceback.format_exc()) self.send_error(400, "Offline fixture assertion failed") def log_message(self, *_args): @@ -205,7 +223,6 @@ def log_message(self, *_args): ) try: outcome = await asyncio.wait_for(execution.run(), 40) - assert not failures assert len(requests) == 2 + triage assert outcome.publications responses = [ @@ -264,3 +281,4 @@ def log_message(self, *_args): server.shutdown() server.server_close() thread.join(timeout=2) + assert not failures, "\n\n".join(failures)