diff --git a/extensions/pi-agent-comms/index.ts b/extensions/pi-agent-comms/index.ts index 04c4550e32..390414df17 100644 --- a/extensions/pi-agent-comms/index.ts +++ b/extensions/pi-agent-comms/index.ts @@ -6,7 +6,7 @@ */ import { execFileSync } from "node:child_process"; -import { readFileSync } from "node:fs"; +import { readFileSync, statSync } from "node:fs"; import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { Type } from "typebox"; @@ -57,13 +57,26 @@ function run(args: string[], launchArguments: string[], timeout = 30_000): Recor export default function (pi: ExtensionAPI) { const currentThread = process.env.AGENT_COMMS_THREAD; pi.registerFlag("comms-catalog", { type: "string", description: "Core native launch catalog" }); - let serviceTier: unknown; + let catalogRevision: string | undefined; + let catalog: Record; + const readCatalog = (): Record => { + const path = pi.getFlag("comms-catalog"); + if (typeof path !== "string") throw new Error("Native launch did not supply its tool catalog"); + const revision = statSync(path, { bigint: true }); + const key = `${revision.mtimeNs}:${revision.size}`; + if (key !== catalogRevision) { + catalog = JSON.parse(readFileSync(path, "utf8")) as Record; + catalogRevision = key; + } + return catalog; + }; let launchArguments: string[]; const forkThread = process.env.PI_PARENT_ID ? currentThread : undefined; const task = process.env.PI_TASK || ""; const managed = process.env.AGENT_COMMS_MANAGED === "1"; pi.on("before_provider_request", (event, ctx) => { if (!currentThread || !ctx.model || !SERVICE_TIER_CAPABLE_APIS.has(ctx.model.api)) return; + const serviceTier = readCatalog().service_tier; if (serviceTier !== "priority") return; if (typeof event.payload !== "object" || event.payload === null || Array.isArray(event.payload)) { ctx.abort(); @@ -75,10 +88,7 @@ export default function (pi: ExtensionAPI) { // standalone forks attach themselves, checkpoint, publish activity and release. // Decode the one launch catalog before any standalone session hook uses it. pi.on("session_start", () => { - const path = pi.getFlag("comms-catalog"); - if (typeof path !== "string") throw new Error("Native launch did not supply its tool catalog"); - const startup = JSON.parse(readFileSync(path, "utf8")) as Record; - serviceTier = startup.service_tier; + const startup = readCatalog(); launchArguments = startup.launch_arguments as string[]; const catalog = startup.tools as ToolDeclaration[]; for (const declaration of catalog) { @@ -88,8 +98,8 @@ export default function (pi: ExtensionAPI) { description: declaration.description, parameters: Type.Unsafe(declaration.parameters), async execute(_id, params, _signal, _onUpdate, ctx) { - if (declaration.name === "comms_model") { - const selected = String(params.model ?? "").trim(); + if (declaration.name === "comms_model" && params.model != null) { + const selected = String(params.model).trim(); const separator = selected.indexOf("/"); if (separator <= 0 || separator === selected.length - 1) { throw new Error("Model must be an exact provider/model identifier"); @@ -116,7 +126,6 @@ export default function (pi: ExtensionAPI) { "tool", ...(currentThread ? ["--from", currentThread] : []), "--tool", declaration.name, "--arguments", JSON.stringify(params), ], launchArguments); - if (declaration.name === "comms_model" && result.thread === currentThread) serviceTier = result.service_tier; return { content: [{ type: "text", text: JSON.stringify(result, null, 2) }], details: result }; }, }); diff --git a/src/agent_comms/channel_coding_tools.py b/src/agent_comms/channel_coding_tools.py index b6611e687b..99b040eb7a 100644 --- a/src/agent_comms/channel_coding_tools.py +++ b/src/agent_comms/channel_coding_tools.py @@ -158,8 +158,8 @@ def launch_arguments(self, package: Path) -> tuple[str, ...]: declarations = NativeToolDeclaration.members_with(NativeToolDeclaration) arguments = ("--tools", ",".join(tool.declared_name for tool in declarations), *NativeToolDeclaration.extension_arguments(package)) - return self.owner.thread.native_launch_arguments( - NativeArguments.parse(arguments), package, self.owner.comms.root, + return self.owner.comms.threads.native_launch_arguments( + self.owner.thread, NativeArguments.parse(arguments), package, ) async def open_transport(self, turn, ownership) -> UnboundToolTransport: diff --git a/src/agent_comms/cli_verbs/configure.py b/src/agent_comms/cli_verbs/configure.py index c743fb416b..0918923b5b 100644 --- a/src/agent_comms/cli_verbs/configure.py +++ b/src/agent_comms/cli_verbs/configure.py @@ -15,12 +15,10 @@ class ConfigureCliCommand( NameArgument, NextTurnConfiguration, CliCommand, declared_name="configure" ): - help = "Set a stopped thread's model, thinking level, or service tier for its next start" + help = "Set a thread's model, thinking level, or service tier for its next turn" def apply(self, ctx: Comms) -> CliResult: thread = ctx.registry.require(self.name) - ctx.registry.status(thread.name).require_inactive( - f"{thread.name} is running; change its settings through its session") asyncio.run(ConfigOptions.configured().validate(self.configured(thread))) updated = ctx.threads.set_thread_configuration(thread, self) return ThreadConfiguredResult(updated.name, updated.model, updated.thinking_level, diff --git a/src/agent_comms/config_options.py b/src/agent_comms/config_options.py index 989c8d818e..a4b86578d1 100644 --- a/src/agent_comms/config_options.py +++ b/src/agent_comms/config_options.py @@ -9,7 +9,7 @@ from collections.abc import Awaitable, Callable, Mapping from contextlib import aclosing from dataclasses import dataclass, field -from functools import cached_property +from functools import cached_property, partial from pathlib import Path from typing import TYPE_CHECKING, ClassVar @@ -23,7 +23,10 @@ from .owner_launch import RetainedOwnerLaunch from .native_arguments import NativeArguments from .native_pi import NativePiRpcLaunch -from .pi_vocabulary import ThinkingChoice +from .pi_vocabulary import ServiceTier, ThinkingChoice +from .coordinator import Coordination +from .field_codec import FieldCodec +from .threads import ChosenServiceTier, NextTurnConfiguration from .thread_model import CatalogSelection, ThreadModel from .pi_commands import ( GetAvailableModels, @@ -56,6 +59,10 @@ class RunningSelection(CatalogSelection): def thinking_level(self) -> type[ThinkingChoice]: return self.thread.thinking_level + @property + def service_tier(self) -> ServiceTier: + return self.thread.service_tier + @property def project_path(self) -> Path: return self.thread.project_path @@ -218,6 +225,28 @@ async def apply( ) +class ServiceTierConfigOption(CatalogConfigOption): + title = "Service tier" + description = "Provider request service tier, applied from this thread's next turn" + category = "service_tier" + + def current_value(self, thread: CatalogSelection) -> str: + return thread.service_tier.value + + async def discover(self, thread: CatalogSelection) -> list[CatalogChoice]: + return [CatalogChoice(name=tier.value.title(), value=tier.value) for tier in ServiceTier] + + async def apply( + self, owner: SessionConfigOptions, thread: Thread, value: str + ) -> None: + await Coordination.run_worker(partial( + owner.comms.threads.set_thread_configuration, thread, + NextTurnConfiguration( + service_tier=ChosenServiceTier(FieldCodec.decode(ServiceTier, value)), + ), + )) + + class ModelCatalog(DeclaredFamily, affix="ModelCatalog"): """The launch's selected discovery source, decoded once at its environment boundary.""" diff --git a/src/agent_comms/pi_native_backend.py b/src/agent_comms/pi_native_backend.py index 5dee3c4899..c730c4824d 100644 --- a/src/agent_comms/pi_native_backend.py +++ b/src/agent_comms/pi_native_backend.py @@ -248,8 +248,8 @@ async def prepare(self, thread: Thread, *, open_native=None) -> StateData: return await open_native( self, self.configuration.launch.binary, - thread.native_launch_arguments( - NativeArguments.parse(self.arguments(thread)), package.path, self.comms.root), + self.comms.threads.native_launch_arguments( + thread, NativeArguments.parse(self.arguments(thread)), package.path), package=package, worktree=thread.worktree.path, root=self.comms.root, @@ -275,8 +275,8 @@ async def run_turn( package = self.comms.owners.require_private_launch().native_package async for event in backend.stream_agent_events( self.configuration.launch.binary, - thread.native_launch_arguments( - NativeArguments.parse(self.arguments(thread)), package.path, self.comms.root), + self.comms.threads.native_launch_arguments( + thread, NativeArguments.parse(self.arguments(thread)), package.path), content.text, delivery.worktree, delivery.environment, diff --git a/src/agent_comms/pi_vocabulary.py b/src/agent_comms/pi_vocabulary.py index ea5736e01c..9f73912221 100644 --- a/src/agent_comms/pi_vocabulary.py +++ b/src/agent_comms/pi_vocabulary.py @@ -219,6 +219,10 @@ class ServiceTier(StrEnum): STANDARD = "standard" PRIORITY = "priority" + @classmethod + def default(cls) -> ServiceTier: + return cls.STANDARD + class ThinkingChoice(ABC): @classmethod diff --git a/src/agent_comms/thread_catalog.py b/src/agent_comms/thread_catalog.py new file mode 100644 index 0000000000..88de5d2af7 --- /dev/null +++ b/src/agent_comms/thread_catalog.py @@ -0,0 +1,33 @@ +"""Publish the native tool catalog from a current registry declaration.""" + +from pathlib import Path + +from .cli_result import ThreadToolCatalogResult +from .field_codec import FieldCodec +from .launch_root import LaunchRoot +from .registry_store import RegistryStore +from .store_files import AtomicFile +from .threads import Thread, ThreadCatalogFile +from .tool_declaration import NativeRequestTools + + +class ThreadCatalog(ThreadCatalogFile): + """The catalog writer owns acquisition through replacement, on every producer.""" + + __slots__ = () + + @classmethod + def publish(cls, thread: Thread, root: Path) -> Path: + launch = LaunchRoot.from_path(root) + with RegistryStore.at(launch.path).current_entry(thread.name) as current: + declared = current.entry.thread + declared.require_original_project(thread) + path = declared.catalog_path(launch.path) + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + catalog = ThreadToolCatalogResult( + tuple(NativeRequestTools.catalog()), declared.service_tier, launch.arguments(), + ) + body = FieldCodec.compact(catalog.to_wire()) + if not path.exists() or path.read_text() != body: + AtomicFile(path).write_text(body) + return path diff --git a/src/agent_comms/thread_management.py b/src/agent_comms/thread_management.py index 46d8ef85eb..f1bcfd79ae 100644 --- a/src/agent_comms/thread_management.py +++ b/src/agent_comms/thread_management.py @@ -49,7 +49,7 @@ from .importing import ImportLimits, ImportedConversation, ThreadImported from .launch_runtime import LaunchRuntime, TagListText from .message_bus import MessageBus -from .native_arguments import ConfiguredBinary +from .native_arguments import ConfiguredBinary, NativeArguments from .native_settings import NativeThreadDefaults from .owner_lifecycle import LaunchInput, OwnerLifecycle, TaskLaunchInput from .pi_vocabulary import NoThinkingOverride, ThinkingChoice, ThinkingLevelField @@ -59,7 +59,9 @@ from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel from .thread_owned_state import ThreadOwnedState from .thread_text import ForkedParent, GivenThreadText, NoThreadText, ThreadText +from .thread_catalog import ThreadCatalog from .threads import KeptModel, ModelChoice, NextTurnConfiguration, Thread +from .tool_declaration import NativeRequestTools from .native_session_reopen import PiSessionId from .tool_result import ToolResult @@ -431,7 +433,18 @@ def configure(thread: Thread) -> Thread: thread.execution.require_native() return configuration.configured(thread) - return self.registry.update_thread(original.name, configure) + updated = self.registry.update_thread(original.name, configure) + ThreadCatalog.publish(updated, self.root) + return updated + + def native_launch_arguments( + self, thread: Thread, arguments: NativeArguments, package: Path, + ) -> tuple[str, ...]: + """Publish the current catalog before launching an extension that consumes it.""" + if not arguments.loads(NativeRequestTools.native_extension(package)): + return arguments.argv + path = ThreadCatalog.publish(thread, self.root) + return (*arguments.argv, "--comms-catalog", str(path)) def initialize_native_configuration( self, name: str, *, model: ThreadModel, thinking_level: type[ThinkingChoice], diff --git a/src/agent_comms/thread_model.py b/src/agent_comms/thread_model.py index e1fc9a05b2..9ea177db25 100644 --- a/src/agent_comms/thread_model.py +++ b/src/agent_comms/thread_model.py @@ -10,7 +10,7 @@ from agent_comms.field_codec import SubclassUnion, boundary_decoder from agent_comms.native_arguments import NativeArguments, SelectedModel from agent_comms.native_failures import NativePiUnavailable -from agent_comms.pi_vocabulary import NoThinkingOverride, ThinkingChoice +from agent_comms.pi_vocabulary import NoThinkingOverride, ServiceTier, ThinkingChoice from agent_comms.thread_text import AbsentText, GivenText, TextChoice if TYPE_CHECKING: @@ -130,3 +130,7 @@ def model(self) -> ModelSelection: @property def thinking_level(self) -> type[ThinkingChoice]: return NoThinkingOverride + + @property + def service_tier(self) -> ServiceTier: + return ServiceTier.default() diff --git a/src/agent_comms/thread_status.py b/src/agent_comms/thread_status.py index 191b9e8334..88f34b14b4 100644 --- a/src/agent_comms/thread_status.py +++ b/src/agent_comms/thread_status.py @@ -30,11 +30,9 @@ from .message_reference import MessageReference from .messaging import Messaging from .turn_item import SourceTurn - from .registration import Registration from .registry_document import RegistryEntry, RegistrySnapshot, RosterVisibility from .thread_identity import GenerationCounter, OwnerIdentity from .threads import Thread - from .tool_caller import ToolCaller from .turn_lease import TurnCompletion, TurnLeaseFence @@ -148,9 +146,6 @@ def starts_owner(self, nxt: ThreadStatus) -> bool: def changes_owner(self, nxt: ThreadStatus) -> bool: return self.starts_owner(nxt) or nxt.starts_owner(self) - def permits_reconfiguration(self, caller: ToolCaller, registry: Registration, name: str) -> bool: - return True - def native_owner_binding(self, snapshot: RegistrySnapshot, thread: Thread) -> ThreadOwnerBinding: return UnavailableThreadOwnerBinding() @@ -240,9 +235,6 @@ def starts_owner(self, nxt: ThreadStatus) -> bool: def withdraw_owner(self, counter: GenerationCounter, name: str) -> None: counter.advance(name) - def permits_reconfiguration(self, caller: ToolCaller, registry: Registration, name: str) -> bool: - return caller.matches_thread(registry, name) - def native_owner_binding(self, snapshot: RegistrySnapshot, thread: Thread) -> ThreadOwnerBinding: if not thread.process_alive: return super().native_owner_binding(snapshot, thread) diff --git a/src/agent_comms/threads.py b/src/agent_comms/threads.py index 1e72ced621..43bde06d3e 100644 --- a/src/agent_comms/threads.py +++ b/src/agent_comms/threads.py @@ -12,7 +12,6 @@ from .agent_backend import AgentBackend, NewThreadBackend, RuntimeLocation from .channel_targets import CommandersTag, DeclaredTag, Tag from .child_process import NoProcess, ProcessIdentity, ProcessState -from .cli_result import ThreadToolCatalogResult from .context_segments.wake_values import WakeWorkContext from .declared_family import DeclaredFamily from .errors import RelationViolationError @@ -32,12 +31,10 @@ GoalDeclaration, NoGoal, ) -from .launch_root import LaunchRoot from .launch_runtime import ThreadLaunch from .lifecycle import LifecycleState from .lifecycle_capability import Executing from .locked_store import DurableFile, PatternFile -from .native_arguments import NativeArguments from .pi_vocabulary import ( NoThinkingOverride, ServiceTier, @@ -51,7 +48,6 @@ ) from .repo import Checkout from .session_file import NoSessionFile, SessionFile -from .store_files import AtomicFile from .task_sources import TaskScope from .thread_creation import CreationSeconds, CreationTime, GeneratedCreationTime from .thread_execution import NativeThreadExecution, ThreadExecution @@ -82,7 +78,6 @@ UnassignedThreadStorage, ) from .thread_text import NoParent, NoThreadText, ThreadParent, ThreadText -from .tool_declaration import NativeRequestTools from .turn_lease import FinishedTurn, NoTurnRecord, TurnLease, TurnLeaseFence, TurnRecord, TurnState if TYPE_CHECKING: @@ -262,8 +257,8 @@ def acting_identity(self) -> ThreadIncarnation: return own.peer_for(own) -class ThreadCatalog(DurableFile, PatternFile): - """The tool catalog a native launch reads from its thread's session directory.""" +class ThreadCatalogFile(DurableFile, PatternFile): + """The catalog's path declaration, usable without importing the registry reader.""" __slots__ = () filename: ClassVar[str] = "native-sessions/{lookup}/comms-catalog.json" @@ -288,7 +283,7 @@ class Thread(ThreadProvenance, CatalogSelection): ThinkingLevelField, RegistrationPolicy(InheritApplied) ] = NoThinkingOverride service_tier: Annotated[ServiceTier, RegistrationPolicy(InheritPrevious)] = ( - field(default=ServiceTier.PRIORITY, kw_only=True) + field(default_factory=ServiceTier.default, kw_only=True) ) goal: Annotated[GoalDeclaration, RegistrationPolicy(GoalDeclaration)] = ( field(default_factory=NoGoal) @@ -439,27 +434,12 @@ def native_environment(self, root, snapshot) -> dict[str, str]: **project.environment(root), } - def native_launch_arguments( - self, arguments: NativeArguments, package: Path, root: Path, - ) -> tuple[str, ...]: - """The existing native session directory owns this root's launch catalog.""" - if not arguments.loads(NativeRequestTools.native_extension(package)): - return arguments.argv - launch = LaunchRoot.from_path(root) - path = self.catalog_path(launch.path) - path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) - catalog = ThreadToolCatalogResult( - tuple(NativeRequestTools.catalog()), self.service_tier, launch.arguments(), - ) - AtomicFile(path).write_text(FieldCodec.compact(catalog.to_wire())) - return (*arguments.argv, "--comms-catalog", str(path)) - def launch(self) -> ThreadLaunch: """The launch environment this thread's owner process carries.""" return ThreadLaunch.from_thread(self) def catalog_path(self, root: Path) -> Path: - return ThreadCatalog.path_in(root, lookup=self.incarnation.require_recorded().lookup) + return ThreadCatalogFile.path_in(root, lookup=self.incarnation.require_recorded().lookup) def native_session_directory(self, root: Path) -> Path: return self.catalog_path(root).parent diff --git a/src/agent_comms/tool_caller.py b/src/agent_comms/tool_caller.py index f78bb2aed5..153b76ee86 100644 --- a/src/agent_comms/tool_caller.py +++ b/src/agent_comms/tool_caller.py @@ -22,7 +22,6 @@ if TYPE_CHECKING: from .comms import Comms - from .registration import Registration from .head_review import HeadReviewOrder, PullRequestHead from .threads import Thread from .tool_request import ToolRequest @@ -87,12 +86,6 @@ def end(self, comms: Comms) -> None: @abstractmethod def as_thread(self, commander: str) -> ThreadCaller: ... - def matches_thread(self, registry: Registration, name: str) -> bool: - return False - - def may_reconfigure(self, comms: Comms, name: str) -> bool: - return comms.registry.status(name).permits_reconfiguration(self, comms.registry, name) - @abstractmethod def require_directive(self, comms: Comms, body: str) -> None: """A caller whose sends direct units refuses a body its directive checks refuse.""" @@ -189,9 +182,6 @@ def from_flag(self) -> tuple[str, ...]: def own_thread(self) -> str: return self.name - def matches_thread(self, registry: Registration, name: str) -> bool: - return registry.canonical_name(name) == registry.canonical_name(self.name) - def require_directive(self, comms: Comms, body: str) -> None: """A unit's or a script's send is a message, not a directive.""" diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index b2e9a429ca..a3058449ff 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -40,7 +40,7 @@ from .cli_result import CliResult from .coordination_tables.refusals import SendRefusal from .display_order import ChannelSort, ThreadSort -from .errors import RelationViolationError, ToolRefusedError +from .errors import ToolRefusedError from .field_codec import FieldCodec, NullMember, SingleFieldValue, SubclassUnion, WireForm, WireName, WireOmitDefault, projected, GoalId from .goal_actions import ActiveGoalAction, EditGoalAction, GoalAction, GoalPrecondition, ModelInvocable, SetGoalAction, TransitionGoalAction from .goal_history import RecordedGoalHistory @@ -340,10 +340,6 @@ def apply(self, comms: Comms, caller: ToolCaller) -> ModelChanged: """Change this thread's model or another thread's model for future turns.""" thread_name = self.thread.name(caller) original = comms.registry.require(thread_name) - if not caller.may_reconfigure(comms, thread_name): - raise RelationViolationError( - f"{thread_name} is running; change its settings through its own session" - ) thread = comms.threads.set_thread_configuration(original, self) return ModelChanged(thread.name, thread.model, thread.thinking_level, thread.service_tier) diff --git a/stack/native/native-import-manifest.json b/stack/native/native-import-manifest.json index d52660d114..aa360090fa 100644 --- a/stack/native/native-import-manifest.json +++ b/stack/native/native-import-manifest.json @@ -18,9 +18,9 @@ "sources": [ { "path": "/home/ts/.pi/agent/extensions/agent-comms/index.ts", - "sha256": "a351953f396227fafbc80abcc1bbba275ec6fd3ef02c18a5f4223be4f713e05e", + "sha256": "5153e0b671b0887ceb8cf8b15b23975d0ff4281672ebb9c46396e02394ee93e5", "snapshot": "extensions/pi-agent-comms/index.ts", - "bytes": 7032 + "bytes": 7285 } ] }, diff --git a/stack/pi-native.sha256 b/stack/pi-native.sha256 index 0ed44a2d97..a38cf50705 100644 --- a/stack/pi-native.sha256 +++ b/stack/pi-native.sha256 @@ -24,8 +24,8 @@ d7ad93bcb341424fa79014ce2fec4226326cfab1e8d5fee27c1859138e120161 dist/core/sdk. 37ae95f1f0ca7c88a75f3c83101e3d7b621c15afced8ac8ec287f958f150cb30 dist/core/native-proof-journal.js bc883f29dbb2a9ba342cb4635a483a5d350ee7d8cb208d43d424449776d41177 dist/core/native-proof-schema.js aae7f609bfa65d29ed16e311456921a88221622e08cf6a1f5c7aab8e58bd69e6 node_modules/@earendil-works/pi-ai/dist/api/openai-completions.js -99f2649a743a3bf2b8765b502c04cd8d45bf0f954bb1419b446669c07de0c4db agent-comms-extensions/global-agent-comms/index.mjs -8b9fce004ea6e178729afa9d2252c5031592125191b9bee796663d2b525c21f5 dist/agent-comms-imports.json +d2f4cbc4d2f2fea8725aa8fe1cf180579a403d6380dde726f2ca506fc79e47b9 agent-comms-extensions/global-agent-comms/index.mjs +3a8ef56ea7f594cf19e1ab62d2354e156b851d7874c9b96951edfccd08091436 dist/agent-comms-imports.json aa9629bcd38a1bbfbde04af47476b36d502084d46677e5532adcdbdf54f8ac1f node_modules/@earendil-works/pi-ai/dist/utils/agent-comms-request-observation.js 1aa5777dcc6e6d98d27a7175e936731d7403173c4971e30adf7c59b076d1d0dd node_modules/@earendil-works/pi-ai/dist/utils/agent-comms-request-observation.d.ts e87b99fc294377e78a6bc6019a6b21816d9ba27012f109fb2626105e0c3f1ceb node_modules/@earendil-works/pi-ai/dist/utils/provider-retry.js @@ -56,5 +56,5 @@ d64e8c05c7932d377034820b1db0df45612afe64885c0765f8f348a12e1bc955 dist/core/skil 8e8276bbabcd898136d431345611928626fad6b6f03ac572f1d8506e6ece326f dist/core/system-prompt.d.ts 0391543cf703a93a0163b37b3ccebe53e0f12bff6f6162b86d56fb017fe49011 dist/core/system-prompt.js # agent-comms-native-stock-v1 @earendil-works/pi-coding-agent@0.85.1 -# agent-comms-native-tree-v1 724dbf68818d1932e6981b737b8e1a8cf4280e592f4d63013306e41e9d15feb8 -# agent-comms-native-inputs-v1 5d32a1691f83443a9c82235830923331f195ce524046354b075d400e08496eb9 +# agent-comms-native-tree-v1 108777c1c49476d1eb75fc6025e1b15e178fe455f8b63a01638fad0359e76567 +# agent-comms-native-inputs-v1 2100e9c81e1789afcd120bc53c9412551f532ea76e9cdb7c1c21cc81fe8e05b0 diff --git a/stack/test-native-global-extensions.py b/stack/test-native-global-extensions.py index ff9ea6cf84..2abd6cc93e 100755 --- a/stack/test-native-global-extensions.py +++ b/stack/test-native-global-extensions.py @@ -114,7 +114,9 @@ def main(fixture: Path): saved_source_sha256 = hashlib.sha256(session.read_bytes()).hexdigest() settings_sha256 = hashlib.sha256((agent / "settings.json").read_bytes()).hexdigest() started = time.monotonic() - arguments = thread.native_launch_arguments(NativeArguments.parse(("--offline",)), package, comms.root) + arguments = comms.threads.native_launch_arguments( + thread, NativeArguments.parse(("--offline",)), package, + ) launch = NativePiRpcLaunch.managed( "pi", arguments, package=verified, worktree=project, root=comms.root, owner=thread.incarnation.require_recorded(), diff --git a/tests/test_acp.py b/tests/test_acp.py index 44abbd441d..b4bcfa8351 100644 --- a/tests/test_acp.py +++ b/tests/test_acp.py @@ -126,6 +126,40 @@ async def test_unknown_model_is_rejected(self, tmp_path, monkeypatch, native_run session_id="proj", config_id="model", value="openrouter/missing" ) + async def test_service_tier_menu_selection_and_readback( + self, tmp_path, native_runtime_package, + ): + from agent_comms.pi_vocabulary import ServiceTier + + agent = self._agent(tmp_path) + try: + response = await agent.new_session(cwd="/wt/proj", mcp_servers=[]) + tier = next(option for option in response.config_options + if option.category == "service_tier") + assert tier.current_value == ServiceTier.STANDARD.value + assert {choice.value for choice in tier.options} == { + member.value for member in ServiceTier + } + original = agent._comms.registry.require(response.session_id) + for selected in (ServiceTier.PRIORITY, ServiceTier.STANDARD): + changed = await agent.set_config_option( + session_id=response.session_id, config_id=tier.id, + value=selected.value, + ) + readback = next(option for option in changed.config_options + if option.id == tier.id) + current = agent._comms.registry.require(response.session_id) + assert readback.current_value == selected.value + assert current.service_tier is selected + assert current.model == original.model + assert current.thinking_level == original.thinking_level + assert current.turn_record == original.turn_record + assert json.loads(current.catalog_path(agent._comms.root).read_text())[ + "service_tier" + ] == selected.value + finally: + await agent.shutdown() + async def test_compaction_details_preserve_markdown(self, tmp_path): summary = "## Decisions\n\n" + "- Keep this decision.\n" * 80 + "\n## Next\nContinue." assert backend.compaction_summary(summary) == summary diff --git a/tests/test_session_components.py b/tests/test_session_components.py index c42a10ed5e..782e2eb4b5 100644 --- a/tests/test_session_components.py +++ b/tests/test_session_components.py @@ -9,17 +9,17 @@ from agent_comms.threads import Thread -def test_thread_service_tier_round_trips_legacy_rows(*, thread_registration): +def test_thread_service_tier_defaults_standard_and_preserves_explicit_priority(*, thread_registration): from agent_comms.field_codec import FieldCodec from agent_comms.pi_vocabulary import ServiceTier encoded = FieldCodec.encode(thread_registration.declare(Thread('legacy', frozenset(), Checkout(Path('.'))))) - assert encoded.pop("service_tier") == "priority" + assert encoded.pop("service_tier") == "standard" restored = FieldCodec.decode(Thread, encoded) - assert restored.service_tier is ServiceTier.PRIORITY - standard = thread_registration.declare(Thread('standard', frozenset(), Checkout(Path('.')), service_tier=ServiceTier.STANDARD)) - assert FieldCodec.decode(Thread, FieldCodec.encode(standard)).service_tier is ServiceTier.STANDARD + assert restored.service_tier is ServiceTier.STANDARD + priority = thread_registration.declare(Thread('priority', frozenset(), Checkout(Path('.')), service_tier=ServiceTier.PRIORITY)) + assert FieldCodec.decode(Thread, FieldCodec.encode(priority)).service_tier is ServiceTier.PRIORITY with pytest.raises(RelationViolationError): - standard.require_original_model_configuration(restored) + priority.require_original_model_configuration(restored) diff --git a/tests/test_thread_catalog_files.py b/tests/test_thread_catalog_files.py index 90afb95310..bdb5cfbfa5 100644 --- a/tests/test_thread_catalog_files.py +++ b/tests/test_thread_catalog_files.py @@ -6,11 +6,11 @@ from agent_comms.locked_store import LockedFile, RootFile from agent_comms.read_ledger import ReadLedger from agent_comms.store_files import store_lock_path -from agent_comms.threads import ThreadCatalog +from agent_comms.threads import ThreadCatalogFile def _catalog(root): - catalog = ThreadCatalog.path_in(root, lookup="0" * 32) + catalog = ThreadCatalogFile.path_in(root, lookup="0" * 32) catalog.parent.mkdir(parents=True) catalog.write_text("{}") return catalog @@ -20,7 +20,7 @@ def test_session_locks_are_undeclared_and_never_locked(tmp_path): catalog = _catalog(tmp_path) lock = store_lock_path(catalog) lock.touch() - assert ThreadCatalog.matching_in(tmp_path) == frozenset((catalog,)) + assert ThreadCatalogFile.matching_in(tmp_path) == frozenset((catalog,)) RootFile.copy_group(lock, tmp_path / "copy") with pytest.raises(ValueError, match="never itself a lockable store"): LockedFile.copy_group(lock, tmp_path / "locked-copy") diff --git a/tests/test_thread_status.py b/tests/test_thread_status.py index 08896fa1dd..047dcae428 100644 --- a/tests/test_thread_status.py +++ b/tests/test_thread_status.py @@ -14,7 +14,6 @@ from agent_comms.activity import Activity, ActivityState from agent_comms.comms import wire from conftest import installed_comms -from agent_comms.cli_verbs.configure import ConfigureCliCommand from agent_comms.child_process import ProcessIdentity from agent_comms.errors import RelationViolationError, UnregisteredThreadError from agent_comms.field_codec import FieldCodec @@ -33,7 +32,7 @@ ThreadStatus, ) from agent_comms.thread_text import GivenThreadText -from agent_comms.tool_caller import NoThreadCaller, ThreadCaller +from agent_comms.tool_caller import ThreadCaller from agent_comms.threads import Thread @@ -199,32 +198,38 @@ def test_presence_owns_roster_membership_and_self_rename(status, listed, rename) status.require_self_rename() -@pytest.mark.parametrize(("status", "allows_other_caller"), [ - (RunningThreadStatus(), False), - (IdleThreadStatus(), False), - (StoppedThreadStatus(), True), - (ArchivedThreadStatus(), True), -]) -def test_reconfiguration_keeps_caller_authority_across_renaming(tmp_path, status, allows_other_caller): +@pytest.mark.parametrize("status", [RunningThreadStatus(), IdleThreadStatus()]) +def test_tier_only_tool_changes_next_catalog_without_changing_active_request(tmp_path, status): + from agent_comms.cli_verbs.tools import ToolsCliCommand + from agent_comms.pi_vocabulary import ServiceTier + from agent_comms.thread_model import DeclaredModel + from agent_comms.tools import invoke_tool + comms = wire(tmp_path) - comms.registry.declare(Thread("subject", frozenset(), Checkout(tmp_path), - process_identity=ProcessIdentity.capture(os.getpid()))) - comms.registry.declare(Thread("other", frozenset(), Checkout(tmp_path))) + comms.registry.declare(Thread("commander", frozenset(), Checkout(tmp_path))) + comms.registry.register(Thread( + "subject", frozenset(), Checkout(tmp_path), + process_identity=ProcessIdentity.capture(os.getpid()), + model=DeclaredModel("openai/test"), service_tier=ServiceTier.PRIORITY, + ), status) comms.registry.rename("subject", "renamed") - comms.registry.register(comms.registry.require("renamed"), status) - assert ThreadCaller("subject").may_reconfigure(comms, "renamed") - assert ThreadCaller("renamed").may_reconfigure(comms, "subject") - assert NoThreadCaller().may_reconfigure(comms, "subject") == allows_other_caller - assert ThreadCaller("other").may_reconfigure(comms, "renamed") == allows_other_caller + original = comms.registry.require("renamed") + active_request = ToolsCliCommand(name=ThreadCaller("renamed")).apply(comms) + result = invoke_tool(comms, "comms_model", { + "thread": "subject", "service_tier": "standard", + }, caller=ThreadCaller("commander")) -@pytest.mark.parametrize("status", [RunningThreadStatus(), IdleThreadStatus()]) -def test_configure_command_keeps_active_configuration_in_its_session(tmp_path, status): - comms = wire(tmp_path) - comms.registry.register(Thread("subject", frozenset(), Checkout(tmp_path), - process_identity=ProcessIdentity.capture(os.getpid())), status) - with pytest.raises(RelationViolationError, match="change its settings through its session"): - ConfigureCliCommand(name="subject").apply(comms) + updated = comms.registry.require("renamed") + next_request = ToolsCliCommand(name=ThreadCaller("renamed")).apply(comms) + assert result["service_tier"] == "standard" + assert updated.service_tier is ServiceTier.STANDARD + assert updated.model == original.model + assert updated.thinking_level == original.thinking_level + assert updated.process_identity == original.process_identity + assert updated.turn_record == original.turn_record + assert active_request.service_tier is ServiceTier.PRIORITY + assert next_request.service_tier is ServiceTier.STANDARD def test_roster_and_controls_keep_archived_threads_dormant(): diff --git a/tests/test_tools.py b/tests/test_tools.py index aed648ac23..1cfac5f173 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -22,6 +22,8 @@ def test_native_launch_carries_the_existing_catalog_and_thread_tier(self, comms, from agent_comms.cli_verbs.tools import ToolsCliCommand from agent_comms.native_arguments import NativeArguments from agent_comms.pi_vocabulary import ServiceTier + from agent_comms.thread_catalog import ThreadCatalog + from agent_comms.threads import ChosenServiceTier, NextTurnConfiguration from agent_comms.tool_caller import ThreadCaller monkeypatch.setenv("AGENT_COMMS_NATIVE_CONFIG_DIR", str(tmp_path / "config")) @@ -30,7 +32,10 @@ def test_native_launch_carries_the_existing_catalog_and_thread_tier(self, comms, name="catalog-owner", tags=frozenset(), worktree=Checkout(Path("/wt")), service_tier=ServiceTier.STANDARD, ) comms.registry.declare(thread) - flag, path = thread.native_launch_arguments(NativeArguments(()), tmp_path / "package", comms.root) + launch_thread = thread + flag, path = comms.threads.native_launch_arguments( + launch_thread, NativeArguments(()), tmp_path / "package", + ) assert flag == "--comms-catalog" payload = json.loads(Path(path).read_text()) assert payload == ToolsCliCommand(name=ThreadCaller(thread.name)).apply(comms).to_wire() @@ -38,6 +43,26 @@ def test_native_launch_carries_the_existing_catalog_and_thread_tier(self, comms, assert {tool["name"] for tool in payload["tools"]} == { tool["name"] for tool in NativeRequestTools.catalog() } + for tier in (ServiceTier.PRIORITY, ServiceTier.STANDARD): + thread = comms.threads.set_thread_configuration( + comms.registry.require(thread.name), + NextTurnConfiguration(service_tier=ChosenServiceTier(tier)), + ) + assert thread.catalog_path(comms.root) == Path(path) + updated = json.loads(Path(path).read_text()) + assert updated == {**payload, "service_tier": tier.value} + revision = Path(path).stat() + # Launch acquired this snapshot before configuration changed. Its + # catalog must still project the current registry tier, not the snapshot. + assert comms.threads.native_launch_arguments( + launch_thread, NativeArguments(()), tmp_path / "package", + ) == (flag, path) + ThreadCatalog.publish(launch_thread, comms.root) + assert json.loads(Path(path).read_text())["service_tier"] == tier.value + repeated = Path(path).stat() + assert (revision.st_mtime_ns, revision.st_size) == ( + repeated.st_mtime_ns, repeated.st_size, + ) def test_collaboration_tools_share_one_mutual_contact(self, comms, monkeypatch): monkeypatch.setenv("AGENT_COMMS_THREAD", "a")