Expose standard-by-default service tier without running-thread locks

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<string, unknown>;
+	const readCatalog = (): Record<string, unknown> => {
+		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<string, unknown>;
+			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<string, unknown>;
-		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 };
 				},
 			});

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:

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,

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."""
 

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,

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

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

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],

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()

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)

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