diff --git a/src/agent_comms/following_inputs.py b/src/agent_comms/following_inputs.py index f23dea0d43..e31077de00 100644 --- a/src/agent_comms/following_inputs.py +++ b/src/agent_comms/following_inputs.py @@ -13,7 +13,7 @@ from .owned_send import OwnedSend from agent_comms.goal_turn_permit import NoLaunchPermit from .queued_input import DeliveredInput, InputHandoffRefused -from .thread_identity import ClaimedIdentity, RegisteredTurn +from .thread_identity import ClaimedIdentity from .turn_input_source import AcceptedFollowingInput if TYPE_CHECKING: @@ -90,7 +90,7 @@ async def open(self, session: TrackedTurnSession) -> None: goal_store=await Coordination.run_worker(inputs.comms.goals.open_goal_store), session_id=self.session_id, thread=thread, - turn=RegisteredTurn(lease.turn_id), + turn=lease.turn_id, claim_generation=lease.claimed_generation, original_source=original, goal_permit=NoLaunchPermit(), diff --git a/src/agent_comms/goal_turn_permit.py b/src/agent_comms/goal_turn_permit.py index 4d825de5b2..34d7b2f44b 100644 --- a/src/agent_comms/goal_turn_permit.py +++ b/src/agent_comms/goal_turn_permit.py @@ -8,12 +8,13 @@ from agent_comms.ordinary_turn_rules import OrdinaryGoalGrantCheck +from .field_codec import TurnId + if TYPE_CHECKING: from agent_comms.goal_attempts import GoalAttemptStore, LaunchPermit from agent_comms.goal_management import Goals from agent_comms.goals import GoalDeclaration from agent_comms.pi_payloads import PiUsage - from agent_comms.thread_identity import RegisteredTurn from agent_comms.threads import Thread from agent_comms.turn_context import TurnContext from agent_comms.turn_goal_account import TurnGoalAccount, VerifiedGoalSettlement @@ -45,7 +46,7 @@ def stop(self, store: GoalAttemptStore) -> None: permit.stop(store) def settle_goal(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptStore, - goal: GoalDeclaration, turn: RegisteredTurn) -> None: + goal: GoalDeclaration, turn: TurnId) -> None: for permit in self.claims: goal.settle(settlement, store, permit, turn) diff --git a/src/agent_comms/goals.py b/src/agent_comms/goals.py index 32dd3e38b8..b0868ff70a 100644 --- a/src/agent_comms/goals.py +++ b/src/agent_comms/goals.py @@ -21,8 +21,15 @@ from .declared_family import DeclaredFamily from .errors import RejectedEnforcementError, RelationViolationError from .field_codec import ( - FieldCodec, JsonShapeFamily, JsonShapeMember, MalformedRecord, NullMember, - PresenceExclude, RecordMember, SubclassUnion, + FieldCodec, + JsonShapeFamily, + JsonShapeMember, + MalformedRecord, + NullMember, + PresenceExclude, + RecordMember, + SubclassUnion, + TurnId, ) from .goal_states import ActiveGoal, GoalState from agent_comms.goal_turn_permit import GoalTurnPermit, NoLaunchPermit @@ -45,7 +52,6 @@ from .goal_management import Goals from .goal_attempts import GoalAttemptStore, LaunchPermit from .turn_goal_account import VerifiedGoalSettlement - from .thread_identity import RegisteredTurn from .todos import Todo @@ -588,7 +594,7 @@ def stopped_permit(self, store: GoalAttemptStore) -> GoalTurnPermit: ... @abstractmethod def settle(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptStore, - permit: LaunchPermit, turn: RegisteredTurn) -> None: ... + permit: LaunchPermit, turn: TurnId) -> None: ... @abstractmethod def instructions_for(self, goals: Goals, owner: ThreadIncarnation) -> tuple[str, ...]: ... @@ -678,7 +684,7 @@ def registration_value(self, previous: GoalDeclaration) -> GoalDeclaration: return previous def settle(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptStore, - permit: LaunchPermit, turn: RegisteredTurn) -> None: + permit: LaunchPermit, turn: TurnId) -> None: pass def instructions_for(self, goals: Goals, owner: ThreadIncarnation) -> tuple[str, ...]: @@ -785,7 +791,7 @@ def registration_value(self, previous: GoalDeclaration) -> GoalDeclaration: return self def settle(self, settlement: type[VerifiedGoalSettlement], store: GoalAttemptStore, - permit: LaunchPermit, turn: RegisteredTurn) -> None: + permit: LaunchPermit, turn: TurnId) -> None: settlement(store, permit, self, turn).dispatch_sync(self.state) def instructions_for(self, goals: Goals, owner: ThreadIncarnation) -> tuple[str, ...]: diff --git a/src/agent_comms/input_disposition.py b/src/agent_comms/input_disposition.py index 003445e96b..3799e2d099 100644 --- a/src/agent_comms/input_disposition.py +++ b/src/agent_comms/input_disposition.py @@ -40,7 +40,6 @@ from .channel_input_batch import SingleInputBatch from .private_registry_guard import RegistryPrivacy from .registry_document import RegistrySnapshot - from .thread_identity import RegisteredTurn from .turn_lease import TurnLeaseFence @@ -299,13 +298,13 @@ def reserve(original: InputDocument) -> DocumentChange[InputDocument, InputDocum return self._update(*_members("key", (row.key for row in originals)), reserve) def reserve_turn( - self, owner: str, turn: RegisteredTurn, claim_scope: int, text: str, + self, owner: str, turn: TurnId, claim_scope: int, text: str, *, ownership: ExitStack, ) -> SingleInputBatch: """Enlist rollback before publication; caller retains the original wire scope.""" from .channel_input_batch import SingleInputBatch - row = ReservedInput(f"turn:{turn.value}", None, owner, claim_scope, owner, text) + row = ReservedInput(f"turn:{turn}", None, owner, claim_scope, owner, text) document = self.reserve_originals(row, ownership=ownership) return SingleInputBatch(document.originals((row.key,))) diff --git a/src/agent_comms/installer/probes/carry.py b/src/agent_comms/installer/probes/carry.py index 193716f14c..6b8b90733b 100644 --- a/src/agent_comms/installer/probes/carry.py +++ b/src/agent_comms/installer/probes/carry.py @@ -90,11 +90,12 @@ def rewrite_stored( ) -> None: """Inside the caller's write transaction, carry stored values of a guarded row: its declared mutation guards are lifted and recreated from the same declaration.""" + (key,) = (field.name for field in row._fields if field.column.primary_key) guards = row.triggers() for name in guards: TriggerKind.drop(db, name) db.executemany( - row._sql(f'UPDATE "{row.declared_name}" SET {{{column.name}}}=? WHERE {{seq}}=?'), + row._sql(f'UPDATE "{row.declared_name}" SET {{{column.name}}}=? WHERE {{{key}}}=?'), updates, ) for sql in guards.values(): diff --git a/src/agent_comms/native_turn_context.py b/src/agent_comms/native_turn_context.py index d1702226d0..6265450889 100644 --- a/src/agent_comms/native_turn_context.py +++ b/src/agent_comms/native_turn_context.py @@ -47,10 +47,9 @@ async def record(self, contexts, prior, thread, lease) -> CapturedContext: """Publish the original SDK observation under its leased owner turn, after the prior this turn holds; the result is the turn's next prior.""" from .coordinator import Coordination - from .thread_identity import RegisteredTurn from .turn_context import RecordedContextTurn - turn = RecordedContextTurn(RegisteredTurn(lease.turn_id), lease.identity) + turn = RecordedContextTurn(lease.turn_id, lease.identity) manifest = self.for_turn(thread.incarnation, turn) return await Coordination.run_worker( partial(contexts.record_after, prior, manifest) diff --git a/src/agent_comms/ordinary_turn_rules.py b/src/agent_comms/ordinary_turn_rules.py index 263c5b077e..21d723c299 100644 --- a/src/agent_comms/ordinary_turn_rules.py +++ b/src/agent_comms/ordinary_turn_rules.py @@ -5,8 +5,8 @@ from dataclasses import dataclass from typing import TYPE_CHECKING +from .field_codec import TurnId from .reservation_rules import ReservationRule, RuleCheck -from .thread_identity import RegisteredTurn if TYPE_CHECKING: from .comms import Comms @@ -39,7 +39,7 @@ class OrdinaryOwnerCheck(RuleCheck): registry: RegistrySnapshot canonical: str claim_generation: int - turn: RegisteredTurn + turn: TurnId class OrdinaryIncarnationRule(ReservationRule): diff --git a/src/agent_comms/owned_send.py b/src/agent_comms/owned_send.py index 2b7edbba9f..6ea73f0604 100644 --- a/src/agent_comms/owned_send.py +++ b/src/agent_comms/owned_send.py @@ -23,10 +23,9 @@ from .coordination_session import NativeBoundary from .coordinator import Coordination from .envelope_claim_transitions import NativeClaim -from .field_codec import InputId +from .field_codec import InputId, TurnId from .input_attempt import InputAlreadyCleared from .reservation_rules import ReservationRule, ReservationViolationError -from .thread_identity import RegisteredTurn from .turn_input_source import ( OriginalTurnInput, TurnInputSource, @@ -53,7 +52,7 @@ class OwnedSend(CodingToolOwner, InputStartListener, SendBoundary): goal_store: GoalAttemptStore session_id: str thread: Thread - turn: RegisteredTurn + turn: TurnId claim_generation: int original_source: OriginalTurnInput goal_permit: GoalTurnPermit @@ -146,7 +145,7 @@ def claim_resource(self, store, turn, resource, slot): owner = snapshot.require(self.owner_name) canonical = resource.normalized(Path(owner.worktree)) scope = NativeClaim(owner_claimed_generation=self.claim_generation, - operation_id=slot, turn_id=self.turn.value, + operation_id=slot, turn_id=self.turn, input_id=turn.command.input_id) return boundary.bus.publisher._publish_acquisition_unlocked( snapshot, owner, canonical, scope, activity=boundary.activity) @@ -170,7 +169,7 @@ def following(self, public_id: InputId, native_id: str, text: str) -> bool: def _started(self, source: TurnInputSource, native_id: str, sent_text: str) -> bool: return source.valid_keys(sent_text) and all( self.inputs.dispositions.started( - key, turn_id=self.turn.value, native_id=native_id, text=sent_text + key, turn_id=self.turn, native_id=native_id, text=sent_text ) for key in source.keys ) diff --git a/src/agent_comms/owned_turn.py b/src/agent_comms/owned_turn.py index 9d2b0f6f76..75b0f0d5e8 100644 --- a/src/agent_comms/owned_turn.py +++ b/src/agent_comms/owned_turn.py @@ -28,7 +28,6 @@ from .native_input_owner import RegistryOwner from .owned_send import OwnedSend from .routing import MessageRoute, ScheduledTurn, TurnRouting -from .thread_identity import RegisteredTurn from .turn_goal_account import TurnGoalAccount from .turn_goal_permission import ( OwnerGoalPermission, @@ -153,7 +152,7 @@ def begin(self, resources: AsyncExitStack, permits: ExitStack) -> bool: origin_claims = ExitStack() permits.callback(origin_claims.close) self.goals = TurnGoalAccount( - comms=self.runner.comms, owner=self.thread, turn=RegisteredTurn(self.turn_id), + comms=self.runner.comms, owner=self.thread, turn=self.turn_id, lease=self.turn_lease, permit=self.goal_permit, pending_origins=self.runner.goals.pending_goal_origins, claims=origin_claims, ) @@ -221,7 +220,7 @@ def prepare_prompt(self, batch: InputBatch): self.relations = self.runner.comms.relationships.context_relations(self.thread.name) self.context = TurnContext.for_owner( self.thread, - RecordedContextTurn(RegisteredTurn(self.turn_id), self.turn_lease.identity), + RecordedContextTurn(self.turn_id, self.turn_lease.identity), self.task, self.runner.comms.views.thread_views(), self.relations, @@ -270,7 +269,7 @@ def reserve_input(self, batch: InputBatch): self.original = original.reserve( self.runner.inputs.dispositions, self.thread, - RegisteredTurn(self.turn_id), + self.turn_id, self.turn_lease.claimed_generation, ownership=reservation, ) @@ -344,7 +343,7 @@ async def stream(self): goal_store=await Coordination.run_worker(self.runner.comms.goals.open_goal_store), session_id=self.session_id, thread=self.thread, - turn=RegisteredTurn(self.turn_id), + turn=self.turn_id, claim_generation=self.turn_lease.claimed_generation, goal_permit=self.goal_permit, original_source=replace(self.original, prompt=rendered.text), diff --git a/src/agent_comms/registration.py b/src/agent_comms/registration.py index 0e3680ae78..fc9bc9ccbc 100644 --- a/src/agent_comms/registration.py +++ b/src/agent_comms/registration.py @@ -21,7 +21,7 @@ from .routing import TurnRouting from .storage_rules import StorageRule from .thread_execution import CommanderForkExecution -from .thread_identity import GenerationCounter, RegisteredTurn +from .thread_identity import GenerationCounter from .thread_status import ( ConnectedThreadStatus, NoBridgeThreadStatus, @@ -326,12 +326,11 @@ def lease_live_turn_with_claim( ) -> tuple[Thread, int]: """Claim a turn against the input generation, allowing unrelated metadata changes.""" Thread.require_declaration(expected) - turn = RegisteredTurn(turn_id) generation = GenerationCounter.require_positive(expected_generation) with self.store.editing() as edit: owner = RegistryOwner.capture_local(edit.original.snapshot, expected.name) owner.require_claim(expected, generation) - leased, _ = edit.lease_turn(owner.thread, turn.value, None) + leased, _ = edit.lease_turn(owner.thread, turn_id, None) edit.commit() return leased, generation @@ -345,14 +344,13 @@ def lease_live_turn_with_generation( ) -> tuple[Thread, int]: """Claim against the exact captured declaration and stable owner generation.""" Thread.require_declaration(expected) - turn = RegisteredTurn(turn_id) generation = GenerationCounter.require_positive(expected_owner_generation) with self.store.editing() as edit: snapshot = edit.original.snapshot owner = RegistryOwner.capture_local(snapshot, expected.name) owner.require_exact(snapshot, expected, generation) owner.thread.turn_record.read().require_idle() - result = edit.lease_turn(owner.thread, turn.value, routing) + result = edit.lease_turn(owner.thread, turn_id, routing) edit.commit() return result @@ -360,11 +358,10 @@ def lease_local_turn( self, name: str, turn_id: TurnId, *, routing: TurnRouting | None = None ) -> tuple[Thread, int]: """Atomic local begin-turn, never reviving a stopped or replaced owner.""" - turn = RegisteredTurn(turn_id) with self.store.editing() as edit: owner = RegistryOwner.capture_local(edit.original.snapshot, name) owner.thread.turn_record.read().require_idle() - result = edit.lease_turn(owner.thread, turn.value, routing) + result = edit.lease_turn(owner.thread, turn_id, routing) edit.commit() return result diff --git a/src/agent_comms/selected_turn.py b/src/agent_comms/selected_turn.py index 20993af9ab..2fc4d89094 100644 --- a/src/agent_comms/selected_turn.py +++ b/src/agent_comms/selected_turn.py @@ -29,7 +29,6 @@ from .wake_candidate_index import WakeCandidateIndex from .context_segments.wake import SelectedWakeSegment from .context_segments.selected import SelectedTriageSegment, SelectedWorkSegment -from .thread_identity import RegisteredTurn from .turn_context import RecordedContextTurn, RenderedInput, TurnContext from .turn_phase import PreparingPhase, PromptAcceptancePhase, PublishingPhase @@ -43,7 +42,7 @@ def context(self, segments) -> TurnContext: lease = owner.require_turn_lease() return TurnContext( owner.incarnation, - RecordedContextTurn(RegisteredTurn(lease.turn_id), lease.identity), + RecordedContextTurn(lease.turn_id, lease.identity), tuple(segments), ) diff --git a/src/agent_comms/task_sources.py b/src/agent_comms/task_sources.py index 71dfb73234..f9c010b159 100644 --- a/src/agent_comms/task_sources.py +++ b/src/agent_comms/task_sources.py @@ -9,14 +9,13 @@ from .channel_targets import CommandTarget from .declared_family import DeclaredFamily from .errors import RelationViolationError -from .field_codec import FieldCodec, WireOmitDefault +from .field_codec import FieldCodec, TurnId, WireOmitDefault from .goals import GoalDeclaration, GoalRevision from .message_reference import MessageReference from .owner_requests import OwnerRequestId, RequestDirective from .projection_source import ProjectionSource from .thread_identity import ( RecordedThreadIncarnation, - RegisteredTurn, ThreadIncarnation, TurnIdentity, UserThreadRole, @@ -406,14 +405,14 @@ def require_correction(self, original: Message, registry: RegistrySnapshot) -> N @dataclass(frozen=True, kw_only=True) class ModelTaskDeclaration(ScopedTaskDeclaration): source_turn: TurnIdentity - source_turn_id: RegisteredTurn + source_turn_id: TurnId permits_agent_revision = True @classmethod def from_claim(cls, owner: Thread, scope: TaskScopeSelection, change: TaskChange, **content) -> Self: lease = owner.require_turn_lease() return cls(scope=scope.select(owner), source_turn=lease.identity, - source_turn_id=RegisteredTurn(lease.turn_id), change=change, **content) + source_turn_id=lease.turn_id, change=change, **content) def __post_init__(self): self.author.require_recorded() @@ -443,7 +442,7 @@ def turn_scope_matches_original(self, previous: ModelTaskDeclaration, registry: def require_emission(self, owner: Thread) -> None: self.require_sender(owner.name) lease = owner.require_turn_lease() - if (lease.identity, lease.turn_id) != (self.source_turn, self.source_turn_id.value): + if (lease.identity, lease.turn_id) != (self.source_turn, self.source_turn_id): raise RelationViolationError("Task declaration's original turn is no longer authorized") self.scope.require_current(owner) diff --git a/src/agent_comms/thread_identity.py b/src/agent_comms/thread_identity.py index b6d614cdb2..6b07dadfef 100644 --- a/src/agent_comms/thread_identity.py +++ b/src/agent_comms/thread_identity.py @@ -442,9 +442,6 @@ def decode(cls, name: str, birth: float) -> ThreadIncarnation: return FieldCodec.decode(cls, {"name": name, "created_at": birth}).incarnation() -@dataclass(frozen=True, slots=True) -class RegisteredTurn(ProjectionSource): - value: TurnId @dataclass(frozen=True, slots=True) diff --git a/src/agent_comms/threads.py b/src/agent_comms/threads.py index 5b8d7a9c17..78b5480fc7 100644 --- a/src/agent_comms/threads.py +++ b/src/agent_comms/threads.py @@ -24,6 +24,7 @@ RegistrationPolicy, SingleFieldValue, SubclassUnion, + TurnId, WireOmitDefault, WireRequired, ) @@ -64,7 +65,6 @@ AgentThreadRole, ObservedTurn, OfferedTools, - RegisteredTurn, ThreadIncarnation, ThreadIncarnationRecord, ThreadRole, @@ -489,7 +489,7 @@ def require_idle(self) -> None: def require_current_turn(self, claim_scope: int) -> None: self.turn_record.read().require_current(claim_scope, self.turn_generation) - def require_turn(self, turn_id: RegisteredTurn, claim_scope: int) -> None: + def require_turn(self, turn_id: TurnId, claim_scope: int) -> None: if not self.turn_lease.attests(claim_scope, turn_id): raise RelationViolationError("live owner does not hold the requested turn") @@ -577,10 +577,10 @@ def turn_lease(self) -> TurnLease: def require_turn_lease(self) -> TurnLeaseFence: return self.turn_lease.require(RelationViolationError("An authorized original turn is required")) - def has_authored_turn(self, identity: TurnIdentity, turn: RegisteredTurn) -> bool: + def has_authored_turn(self, identity: TurnIdentity, turn: TurnId) -> bool: return self.turn_lease.authored(identity, turn) - def has_observed_task_turn(self, identity: TurnIdentity, turn: RegisteredTurn) -> bool: + def has_observed_task_turn(self, identity: TurnIdentity, turn: TurnId) -> bool: """Check whether a task record can refer to this thread's current or preceding turn. Reject a different thread registration. During a turn, accept its exact @@ -593,7 +593,7 @@ def has_observed_task_turn(self, identity: TurnIdentity, turn: RegisteredTurn) - return self.turn_lease.observes_task_turn( identity, turn, idle=identity.generation == self.turn_generation - and self.turn_record == FinishedTurn(turn.value)) + and self.turn_record == FinishedTurn(turn)) @property def turn_identity(self) -> TurnIdentity: diff --git a/src/agent_comms/turn_context.py b/src/agent_comms/turn_context.py index 5d059ebe0c..21d8e29841 100644 --- a/src/agent_comms/turn_context.py +++ b/src/agent_comms/turn_context.py @@ -20,18 +20,19 @@ from .declared_family import DeclaredFamily from .field_codec import ( FieldCodec, + IdentityExclude, NullMember, Sha256Text, SingleFieldValue, SubclassUnion, - IdentityExclude, + TurnId, WireOmitDefault, ) from .goals import Goal, NoGoal from .message_reference import MessageReference from .native_session_reopen import NativeSessionIdentity from .peer_relevance import AlwaysListed, GoalShown, PeerRelevance -from .thread_identity import ThreadIncarnation, RegisteredTurn, TurnIdentity +from .thread_identity import ThreadIncarnation, TurnIdentity from .tool_result import DeclaredReply if TYPE_CHECKING: @@ -1194,7 +1195,7 @@ def matches_generation(self, generation: int) -> bool: @dataclass(frozen=True) class RecordedContextTurn(ContextTurn): - identity: RegisteredTurn + identity: TurnId occurrence: TurnIdentity def __post_init__(self): @@ -1202,7 +1203,7 @@ def __post_init__(self): raise ValueError("Context requires the original authorized turn generation") def source_revision(self, owner: Thread) -> str: - return self.identity.value + return self.identity def require_recorded(self) -> RecordedContextTurn: return self diff --git a/src/agent_comms/turn_goal_account.py b/src/agent_comms/turn_goal_account.py index ed8bdb835f..2820e43c38 100644 --- a/src/agent_comms/turn_goal_account.py +++ b/src/agent_comms/turn_goal_account.py @@ -13,6 +13,7 @@ from agent_comms.coordination_errors import RequestRefused +from .field_codec import TurnId from agent_comms.goal_turn_permit import GoalTurnPermit, NoLaunchPermit from .goal_states import ActiveGoal, CompletedGoal, PausedGoal from .mro_dispatch import MroDispatch, handles @@ -24,7 +25,6 @@ from .comms import Comms from .goals import Goal, GoalDeclaration from .pi_payloads import PiUsage - from .thread_identity import RegisteredTurn from .threads import Thread from .turn_lease import TurnLeaseFence @@ -34,7 +34,7 @@ class VerifiedGoalSettlement(MroDispatch): store: GoalAttemptStore permit: LaunchPermit goal: Goal - turn: RegisteredTurn + turn: TurnId @classmethod def settle_current(cls, account: TurnGoalAccount) -> None: @@ -47,12 +47,12 @@ def settle_current(cls, account: TurnGoalAccount) -> None: store = comms.goals.open_goal_store() generation = store.require_generation(goal.id) generation.lifecycle.continue_work( - store, generation, f"verified-turn:{account.turn.value}" + store, generation, f"verified-turn:{account.turn}" ) @property def progress_witness(self) -> str: - return f"native-terminal:{self.turn.value}" + return f"native-terminal:{self.turn}" @property def completion_witness(self) -> str: @@ -65,14 +65,14 @@ def progress(self, state) -> None: @handles(CompletedGoal) def completion(self, state: CompletedGoal) -> None: - if self.goal.reports_turn(self.turn.value): + if self.goal.reports_turn(self.turn): self.permit.record_verified_completion(self.store, self.completion_witness) class OriginGoalSettlement(VerifiedGoalSettlement): @property def progress_witness(self) -> str: - return f"origin-final:{self.turn.value}" + return f"origin-final:{self.turn}" @property def completion_witness(self) -> str: @@ -83,7 +83,7 @@ def completion_witness(self) -> str: class TurnGoalAccount: comms: Comms owner: Thread - turn: RegisteredTurn + turn: TurnId lease: TurnLeaseFence permit: GoalTurnPermit pending_origins: dict[str, str] diff --git a/src/agent_comms/turn_input_source.py b/src/agent_comms/turn_input_source.py index 74d1d1cee4..27e493040e 100644 --- a/src/agent_comms/turn_input_source.py +++ b/src/agent_comms/turn_input_source.py @@ -14,7 +14,7 @@ from .acp_extension import InputStartedUpdate, NoQueueScope from .agent_backend import BusyInputPolicy, SteerInputPolicy from .channel_input_batch import InputBatch, SingleInputBatch -from .field_codec import InputId +from .field_codec import InputId, TurnId from .input_attempt import InputAlreadyCleared from .messages import Message from .ordinary_turn_rules import ( @@ -38,7 +38,7 @@ from .input_drain import DrainedSession, InputDrain from .owned_send import OwnedSend from .runtime import RuntimeServer - from .thread_identity import RecordedThreadIncarnation, RegisteredTurn + from .thread_identity import RecordedThreadIncarnation from .threads import Thread @@ -173,7 +173,7 @@ def _prepare( if not sender.inputs.dispositions.bind_turn_input( self.keys, claim_scope=sender.claim_generation, - turn_id=sender.turn.value, + turn_id=sender.turn, native_id=native_id, text=sent_text, already_bound=already_bound, @@ -278,7 +278,7 @@ def valid_keys(self, text: str) -> bool: return super().valid_keys(text) or (self.batch.allows_multiple and text == self.prompt) def reserve( - self, dispositions: InputDispositions, owner: Thread, turn: RegisteredTurn, + self, dispositions: InputDispositions, owner: Thread, turn: TurnId, claim_scope: int, *, ownership: ExitStack, ) -> OriginalTurnInput: diff --git a/src/agent_comms/turn_lease.py b/src/agent_comms/turn_lease.py index 8c687049a7..26590ef206 100644 --- a/src/agent_comms/turn_lease.py +++ b/src/agent_comms/turn_lease.py @@ -29,7 +29,6 @@ from .thread_identity import ( NoObservedTurn, ObservedTurn, - RegisteredTurn, ThreadIncarnation, TurnFence, TurnIdentity, @@ -75,7 +74,7 @@ def require_idle(self) -> None: ... def require_current(self, claim_scope: int, generation: int) -> None: ... @abstractmethod - def require_turn(self, turn_id: RegisteredTurn) -> None: ... + def require_turn(self, turn_id: TurnId) -> None: ... @abstractmethod def without_claim(self) -> TurnDeclaration: ... @@ -92,7 +91,7 @@ def started_by(self, updated_at_ms: int) -> bool: ... def phase_effects(self, phase: TurnPhase) -> Iterator[TurnDeclaration]: ... @abstractmethod - def matches(self, turn_id: RegisteredTurn) -> bool: ... + def matches(self, turn_id: TurnId) -> bool: ... @dataclass(frozen=True) @@ -120,7 +119,7 @@ def require_idle(self) -> None: def require_current(self, claim_scope: int, generation: int) -> None: pass - def require_turn(self, turn_id: RegisteredTurn) -> None: + def require_turn(self, turn_id: TurnId) -> None: raise RelationViolationError("live owner does not hold the requested turn") def without_claim(self) -> NoTurn: @@ -137,7 +136,7 @@ def started_by(self, updated_at_ms: int) -> bool: def phase_effects(self, phase: TurnPhase) -> Iterator[TurnDeclaration]: return iter(()) - def matches(self, turn_id: RegisteredTurn) -> bool: + def matches(self, turn_id: TurnId) -> bool: return False @@ -254,12 +253,12 @@ def require_current(self, claim_scope: int, generation: int) -> None: if not self.current(claim_scope, generation): raise RelationViolationError("live owner turn claim_scope is no longer current") - def require_turn(self, turn_id: RegisteredTurn) -> None: + def require_turn(self, turn_id: TurnId) -> None: if not self.matches(turn_id): raise RelationViolationError("live owner does not hold the requested turn") - def matches(self, turn_id: RegisteredTurn) -> bool: - return RegisteredTurn(self.id) == turn_id + def matches(self, turn_id: TurnId) -> bool: + return self.id == turn_id def without_claim(self) -> ActiveTurn: return replace(self, claimed_generation=UnrecordedGeneration()) @@ -322,14 +321,14 @@ def require(self, refusal: Exception) -> TurnLeaseFence: ... def held(self) -> tuple[TurnLeaseFence, ...]: ... @abstractmethod - def attests(self, claim_scope: int, turn_id: RegisteredTurn) -> bool: + def attests(self, claim_scope: int, turn_id: TurnId) -> bool: """This lease holds ``turn_id`` under ``claim_scope``.""" @abstractmethod - def authored(self, identity: TurnIdentity, turn: RegisteredTurn) -> bool: ... + def authored(self, identity: TurnIdentity, turn: TurnId) -> bool: ... @abstractmethod - def observes_task_turn(self, identity: TurnIdentity, turn: RegisteredTurn, idle: bool) -> bool: + def observes_task_turn(self, identity: TurnIdentity, turn: TurnId, idle: bool) -> bool: """During a turn, its exact ID or the preceding generation; idle, the caller's answer.""" @abstractmethod @@ -360,13 +359,13 @@ def require(self, refusal: Exception) -> TurnLeaseFence: def held(self) -> tuple[TurnLeaseFence, ...]: return () - def attests(self, claim_scope: int, turn_id: RegisteredTurn) -> bool: + def attests(self, claim_scope: int, turn_id: TurnId) -> bool: return False - def authored(self, identity: TurnIdentity, turn: RegisteredTurn) -> bool: + def authored(self, identity: TurnIdentity, turn: TurnId) -> bool: return False - def observes_task_turn(self, identity: TurnIdentity, turn: RegisteredTurn, idle: bool) -> bool: + def observes_task_turn(self, identity: TurnIdentity, turn: TurnId, idle: bool) -> bool: return idle @boundary_decoder @@ -395,13 +394,13 @@ def require(self, refusal: Exception) -> TurnLeaseFence: def held(self) -> tuple[TurnLeaseFence, ...]: return (self,) - def attests(self, claim_scope: int, turn_id: RegisteredTurn) -> bool: - return self.can_attest(claim_scope) and RegisteredTurn(self.turn_id) == turn_id + def attests(self, claim_scope: int, turn_id: TurnId) -> bool: + return self.can_attest(claim_scope) and self.turn_id == turn_id - def authored(self, identity: TurnIdentity, turn: RegisteredTurn) -> bool: - return (self.identity, self.turn_id) == (identity, turn.value) + def authored(self, identity: TurnIdentity, turn: TurnId) -> bool: + return (self.identity, self.turn_id) == (identity, turn) - def observes_task_turn(self, identity: TurnIdentity, turn: RegisteredTurn, idle: bool) -> bool: + def observes_task_turn(self, identity: TurnIdentity, turn: TurnId, idle: bool) -> bool: return self.authored(identity, turn) or identity.generation == self.identity.generation - 1 @boundary_decoder diff --git a/src/agent_comms/turn_terminal.py b/src/agent_comms/turn_terminal.py index c10fd86ddd..245dd3fd5b 100644 --- a/src/agent_comms/turn_terminal.py +++ b/src/agent_comms/turn_terminal.py @@ -64,7 +64,7 @@ def failure_observation(self, account, permit, captured): owner=account.owner, goal=captured, lease=account.lease, - turn_id=account.turn.value, + turn_id=account.turn, claim_scope=account.lease.claimed_generation, current_owner=snapshot.threads.get(account.thread_name), current_claim=snapshot.claim_generation(account.thread_name), diff --git a/tests/test_carry_context_turn_ids.py b/tests/test_carry_context_turn_ids.py new file mode 100644 index 0000000000..69c880c845 --- /dev/null +++ b/tests/test_carry_context_turn_ids.py @@ -0,0 +1,42 @@ +"""The context carry unboxes stored identities once, before the normal reader.""" + +from cutover.carry_context_turn_ids import ContextTurnIds + +from agent_comms.context_store import ContextRequest, ContextStore, ContextStoreInfo +from agent_comms.field_codec import FieldCodec, TurnId +from agent_comms.installer.probes.carry import Carried +from agent_comms.thread_identity import RecordedThreadIncarnation, TurnIdentity +from agent_comms.turn_context import RecordedContextIdentity, RecordedContextTurn + + +def test_context_turn_ids_unboxes_stored_identity_once(tmp_path): + root = tmp_path / "root" + root.mkdir(mode=0o700) + store = ContextStore(root) + turn_id = TurnId.mint() + owner = RecordedThreadIncarnation("owner", 1.0) + turn = RecordedContextTurn(turn_id, TurnIdentity(owner, 1)) + identity = RecordedContextIdentity(owner, turn, "pi.estimateTokens") + old = FieldCodec.encode(identity) + old["turn"]["identity"] = {"value": FieldCodec.encode(turn_id)} + table = ContextRequest + with store.writing() as db: + ContextStoreInfo.source_in(db).initialized(db) + db.execute( + table._sql( + f'INSERT INTO "{table.declared_name}" ({{identity}}, {{versions}}, {{size}}) ' + "VALUES (?, ?, ?)" + ), + (FieldCodec.compact(old), FieldCodec.compact(()), 0), + ) + + step = ContextTurnIds(root, root, tmp_path / "spill") + assert step.carry(apply=False) == Carried(1) + carried = step.carry(apply=True) + assert carried.changed == 1 + assert carried.rewritten == (ContextStore.path_in(root).relative_to(root),) + with store.snapshot() as db: + recorded = ContextRequest.require(db, id=1).identity.turn.require_recorded() + assert recorded == turn + assert recorded.identity == turn_id + assert step.carry(apply=True) == Carried(0) diff --git a/tests/test_claim_bus_read.py b/tests/test_claim_bus_read.py index 473653fc98..8610328095 100644 --- a/tests/test_claim_bus_read.py +++ b/tests/test_claim_bus_read.py @@ -41,7 +41,7 @@ from agent_comms.owner_fence import prepare_fence_token from agent_comms.read_basis import AllChannelDisplayScope, DMDisplayScope from agent_comms.selected_source_batch import SelectedSource -from agent_comms.thread_identity import RegisteredTurn, TurnIdentity +from agent_comms.thread_identity import TurnIdentity from agent_comms.turn_context import ( ContextManifest, OwnerProvenance, @@ -70,7 +70,7 @@ def record_context(comms: Comms, name: str, generation: int) -> None: ContextManifest( owner.incarnation, RecordedContextTurn( - RegisteredTurn(TurnId(f"{generation:032x}")), + TurnId(f"{generation:032x}"), TurnIdentity(owner.incarnation, generation), ), (segment,), diff --git a/tests/test_context_manifest_deltas.py b/tests/test_context_manifest_deltas.py index 990ccbcddd..48604821bd 100644 --- a/tests/test_context_manifest_deltas.py +++ b/tests/test_context_manifest_deltas.py @@ -11,7 +11,7 @@ from agent_comms.field_codec import FieldCodec, TurnId from agent_comms.native_session_reopen import NativeSessionIdentity -from agent_comms.thread_identity import RegisteredTurn, TurnIdentity +from agent_comms.thread_identity import TurnIdentity from agent_comms.turn_context import ObservedDispatchRequest from agent_comms.turn_context import ( ContextManifest, JournalProvenance, NativeProvenance, RecordedContextPredecessor, @@ -36,7 +36,7 @@ def segment(kind, text, sources=provenance, contributors=()): return ContextManifest( owner.incarnation, RecordedContextTurn( - RegisteredTurn(TurnId(f"{turn:032x}")), TurnIdentity(owner.incarnation, turn) + TurnId(f"{turn:032x}"), TurnIdentity(owner.incarnation, turn) ), ( segment(SystemLayerSegment, 's' * 27000), @@ -60,7 +60,7 @@ def test_context_deltas_restore_original_text_and_incremental_membership(tmp_pat replace( second, turn=replace( - second.turn, identity=RegisteredTurn(TurnId('431ad49b3e0129a11ea5a0320c188024')) + second.turn, identity=TurnId('431ad49b3e0129a11ea5a0320c188024') ), request_id=ObservedDispatchRequest('request-3'), ), diff --git a/tests/test_context_source_inspection.py b/tests/test_context_source_inspection.py index f6bcfa021c..9ff6b46311 100644 --- a/tests/test_context_source_inspection.py +++ b/tests/test_context_source_inspection.py @@ -14,12 +14,20 @@ from agent_comms.native_turn_context import NativeContextData, NativeContextManifestData from agent_comms.runtime_requests import (RuntimeRequest, ContextRecordedSegmentRuntimeRequest, ContextCoreRuntimeRequest, ContextCoreSourceRuntimeRequest) -from agent_comms.thread_identity import RegisteredTurn, TurnIdentity +from agent_comms.thread_identity import TurnIdentity from agent_comms.thread_identity import RecordedThreadIncarnation, ReportedTools from agent_comms.turn_context import ObservedDispatchRequest, UnobservedDispatchRequest from agent_comms.turn_context import ( - ContextManifest, ContextSourceText, FileProvenance, JournalProvenance, - NativeProvenance, PreviewProvenance, RecordedContextTurn, SystemLayerSegment, TranscriptSegment, TurnContext, + ContextManifest, + ContextSourceText, + FileProvenance, + JournalProvenance, + NativeProvenance, + PreviewProvenance, + RecordedContextTurn, + SystemLayerSegment, + TranscriptSegment, + TurnContext, ) from bus_record_fixture import context_comms @@ -37,7 +45,7 @@ def identity(tmp_path): def recorded(segments): owner = RecordedThreadIncarnation("original-owner", 17002.0) turn = RecordedContextTurn( - RegisteredTurn(TurnId('ab87f599dc0a50a56b7c65de71c39426')), TurnIdentity(owner, 2) + TurnId('ab87f599dc0a50a56b7c65de71c39426'), TurnIdentity(owner, 2) ) return ContextManifest(owner, turn, tuple(segments), "pi.estimateTokens", ObservedDispatchRequest("request-one")) diff --git a/tests/test_goal_answers.py b/tests/test_goal_answers.py index bba3e494dd..1c77dce995 100644 --- a/tests/test_goal_answers.py +++ b/tests/test_goal_answers.py @@ -13,7 +13,6 @@ from agent_comms.goal_turn_permit import NoLaunchPermit from agent_comms.goals import NoGoal from agent_comms.pi_payloads import PiUsage -from agent_comms.thread_identity import RegisteredTurn from agent_comms.turn_goal_account import TurnGoalAccount from agent_comms.turn_terminal import EndedWithoutResult, OperatorStopped from agent_comms.threads import Thread @@ -27,8 +26,8 @@ def goal_turn(comms, original_goal): )) if original_goal: goal = comms.goals.update_goal("worker", SetGoalAction(text="Original work")) - turn = RegisteredTurn(TurnId.mint()) - owner = comms.agents.begin_turn("worker", turn.value) + turn = TurnId.mint() + owner = comms.agents.begin_turn("worker", turn) lease = owner.thread.require_turn_lease() try: with ExitStack() as claims: diff --git a/tests/test_input_documents.py b/tests/test_input_documents.py index 716bfb3531..290460e8c5 100644 --- a/tests/test_input_documents.py +++ b/tests/test_input_documents.py @@ -21,7 +21,6 @@ ) from agent_comms.input_disposition import InputDispositionRow, InputDispositions, InputDocument from agent_comms.errors import RelationViolationError -from agent_comms.thread_identity import RegisteredTurn from agent_comms.thread_identity import RecordedThreadIncarnation from stored_inputs import insert_inputs, stored_inputs from agent_comms.threads import Thread @@ -81,7 +80,7 @@ def test_failure_after_original_publication_retires_the_batch_and_schedule(tmp_p with pytest.raises(OSError), ExitStack() as ownership: store.reserve_turn( "worker", - RegisteredTurn(TurnId('6aef76c6cb359f6ffaf9c6e9881c902b')), + TurnId('6aef76c6cb359f6ffaf9c6e9881c902b'), 1, "Original schedule", ownership=ownership, @@ -92,7 +91,7 @@ def test_failure_after_original_publication_retires_the_batch_and_schedule(tmp_p with ExitStack() as ownership, pytest.raises(RelationViolationError): store.reserve_turn( "worker", - RegisteredTurn(TurnId('6aef76c6cb359f6ffaf9c6e9881c902b')), + TurnId('6aef76c6cb359f6ffaf9c6e9881c902b'), 1, "Replacement", ownership=ownership, diff --git a/tests/test_owned_turn_goal_replacement.py b/tests/test_owned_turn_goal_replacement.py index 8041dd50a7..4b46b3e14d 100644 --- a/tests/test_owned_turn_goal_replacement.py +++ b/tests/test_owned_turn_goal_replacement.py @@ -75,7 +75,7 @@ async def test_turn_after_order_replacement_is_allowed(tmp_path, native_runtime_ await wait_for(lambda: saved.provider.posts == 1) sender = backend.TurnSession.active[task].send_boundary assert isinstance(sender, OwnedSend) - original_turn = sender.turn.value + original_turn = sender.turn original_process = comms.registry.require(worker).process_identity todo = comms.goals.todos.get('P1') replaced = comms.goals.update_goal(worker, OrderGoalAction( diff --git a/tests/test_task_decisions.py b/tests/test_task_decisions.py index 114dd9759e..a83b9bd2af 100644 --- a/tests/test_task_decisions.py +++ b/tests/test_task_decisions.py @@ -19,7 +19,6 @@ from agent_comms.goals import Goal from agent_comms.messages import Message, InfoKind from agent_comms.task_sources import Decision -from agent_comms.thread_identity import RegisteredTurn from agent_comms.tools import invoke_tool from agent_comms.tool_caller import ThreadCaller from agent_comms.turn_lease import NoTurnLease @@ -298,7 +297,7 @@ def test_stale_goal_scope_cannot_publish_even_through_original_publisher(comms): declaration = Decision( chosen="A", rejected=("B",), scope=owner.task_scope, source_turn=owner.require_turn_lease().identity, - source_turn_id=RegisteredTurn(owner.require_turn_lease().turn_id), + source_turn_id=owner.require_turn_lease().turn_id, ) comms.registry.register(replace(owner, goal=replace(owner.goal, revision=2))) message = Message("alpha", "#team", declaration.text, InfoKind, task=declaration) diff --git a/tests/test_turn_context_wire.py b/tests/test_turn_context_wire.py index ae084d8eda..13c501315f 100644 --- a/tests/test_turn_context_wire.py +++ b/tests/test_turn_context_wire.py @@ -6,7 +6,7 @@ from agent_comms.image_input import ImageInput from agent_comms.native_tool_call import OwnerToolConnection from agent_comms.pi_commands import Prompt -from agent_comms.thread_identity import RegisteredTurn, TurnIdentity +from agent_comms.thread_identity import TurnIdentity from agent_comms.turn_context import ( ContextManifest, OwnerProvenance, @@ -26,7 +26,7 @@ def manifest(owner, generation=1): return ContextManifest( owner.incarnation, RecordedContextTurn( - RegisteredTurn(TurnId('ab87f599dc0a50a56b7c65de71c39426')), + TurnId('ab87f599dc0a50a56b7c65de71c39426'), TurnIdentity(owner.incarnation, generation), ), (segment,), diff --git a/tests/test_turn_declaration.py b/tests/test_turn_declaration.py index 916ee8bc92..4d1e01a099 100644 --- a/tests/test_turn_declaration.py +++ b/tests/test_turn_declaration.py @@ -8,7 +8,6 @@ from agent_comms.thread_identity import ( NoObservedTurn, RecordedThreadIncarnation, - RegisteredTurn, TurnFence, TurnIdentity, ) @@ -32,7 +31,7 @@ def test_idle_turn_answers_without_an_active_record(): assert FieldCodec.encode(idle) is None idle.require_idle() assert idle.report_turn == '' - assert not idle.matches(RegisteredTurn(TurnId('9b473845f684444f80a26586ae298d2b'))) + assert not idle.matches(TurnId('9b473845f684444f80a26586ae298d2b')) assert not idle.started_by(10000) identity = TurnIdentity(RecordedThreadIncarnation("owner", 1.0), 1) turn_id = TurnId.mint() @@ -50,6 +49,6 @@ def test_idle_turn_answers_without_an_active_record(): with pytest.raises(RelationViolationError): idle.require_active() with pytest.raises(RelationViolationError): - idle.require_turn(RegisteredTurn(TurnId('9b473845f684444f80a26586ae298d2b'))) + idle.require_turn(TurnId('9b473845f684444f80a26586ae298d2b')) diff --git a/tests/test_turn_goal_account.py b/tests/test_turn_goal_account.py index 76d0cbe882..5ddac5ba6b 100644 --- a/tests/test_turn_goal_account.py +++ b/tests/test_turn_goal_account.py @@ -11,7 +11,6 @@ ) from agent_comms.goal_states import ActiveGoal, CompletedGoal, PausedGoal from agent_comms.goals import Goal, ReportedTurn -from agent_comms.thread_identity import RegisteredTurn from agent_comms.turn_goal_account import OriginGoalSettlement, VerifiedGoalSettlement @@ -22,14 +21,14 @@ class SpecializedPausedGoal(PausedGoal): (tmp_path / "attempts").mkdir(mode=0o700) store = GoalAttemptStore.initialize(tmp_path / "attempts") - turn = RegisteredTurn(TurnId('9b473845f684444f80a26586ae298d2b')) + turn = TurnId('9b473845f684444f80a26586ae298d2b') for settlement in (VerifiedGoalSettlement, OriginGoalSettlement): for state in (ActiveGoal(), PausedGoal(), CompletedGoal(), SpecializedPausedGoal()): goal = Goal( "work", f"{settlement.__name__}-{state.declared_name}", published_seq=0, - reported_turn=ReportedTurn(turn.value), + reported_turn=ReportedTurn(turn), state=state, ) store.create_goal(goal.id) diff --git a/tools/cutover/carry_context_turn_ids.py b/tools/cutover/carry_context_turn_ids.py new file mode 100644 index 0000000000..fff609ee66 --- /dev/null +++ b/tools/cutover/carry_context_turn_ids.py @@ -0,0 +1,96 @@ +"""Unbox recorded context turn identities before the nominal reader opens them.""" + +from abc import ABC, abstractmethod +from contextlib import closing +from dataclasses import dataclass +from functools import partial +from pathlib import Path +from typing import ClassVar + +from agent_comms.context_store import ContextRequest, ContextStore +from agent_comms.declared_family import DeclaredFamily +from agent_comms.field_codec import FieldCodec, MalformedRecord, TurnId +from agent_comms.installer.probes.carry import Carried, Carry +from agent_comms.locked_store import RootFile +from agent_comms.thread_identity import TurnIdentity +from agent_comms.turn_context import RecordedContextTurn + + +@dataclass(frozen=True) +class StoredRegisteredTurn: + value: TurnId + + +class CarriedContextTurn(ABC): + @abstractmethod + def carried(self) -> RecordedContextTurn: ... + + +class BoxedContextTurn(DeclaredFamily, CarriedContextTurn, affix="BoxedContextTurn"): ... + + +class UnboxedContextTurn(DeclaredFamily, CarriedContextTurn, affix="UnboxedContextTurn"): ... + + +@dataclass(frozen=True) +class RecordedBoxedContextTurn(BoxedContextTurn): + identity: StoredRegisteredTurn + occurrence: TurnIdentity + + def carried(self) -> RecordedContextTurn: + return RecordedContextTurn(self.identity.value, self.occurrence) + + +@dataclass(frozen=True) +class RecordedUnboxedContextTurn(UnboxedContextTurn): + identity: TurnId + occurrence: TurnIdentity + + def carried(self) -> RecordedContextTurn: + return RecordedContextTurn(self.identity, self.occurrence) + + +class ContextTurnIds(Carry): + opened_files: ClassVar[tuple[type[RootFile], ...]] = (ContextStore,) + + @staticmethod + def rewrite(path: Path, record: type, payload: dict[str, object]) -> dict[str, object]: + try: + return { + **FieldCodec.encode( + FieldCodec.decode( + RecordedBoxedContextTurn | RecordedUnboxedContextTurn, payload + ).carried() + ) + } + except MalformedRecord as error: + error.add_note(f"{path}: {payload!r}") + raise + + def carry(self, *, apply: bool) -> Carried: + table = ContextRequest + (column,) = table._fields_named(("identity",)) + changed = 0 + rewritten: list[Path] = [] + for path in self.input_files(): + updates: list[tuple[object, int]] = [] + with closing(ContextStore.read_snapshot(path)) as db: + if not table.present(db): + continue + for key, stored in db.execute( + table._sql(f'SELECT {{id}}, {{identity}} FROM "{table.declared_name}"') + ): + original = column.storage.from_sql(stored) + current = column.value.rewrite_document( + original, + partial(self.rewrite, path), + lambda record: record is RecordedContextTurn, + ) + if current != original: + updates.append((column.storage.to_sql(current), key)) + changed += len(updates) + if apply and updates: + with closing(ContextStore.writer(path)) as db, ContextStore.write_transaction(db): + self.rewrite_stored(db, table, column, updates) + rewritten.append(path.relative_to(self.root)) + return Carried(changed, rewritten=tuple(rewritten))