refactor: framework review 2026-08 — tool identity, lifecycle, sessions, jobs

CLAUDE.md

@@ -159,7 +159,15 @@ Workflow:
 - **Tool registration**: Use `@register_tool(category=..., capabilities=..., description=...)` decorator. `capabilities=` is required — pass `EXEMPT` for tools that need no permission check or a list of `Capability(name, target_arg=...)` tuples the engine matches against rules. Tools are auto-discovered via the global `ToolRegistry`.
 - **Permissions**: `workflow/permissions/` holds a framework-independent engine that evaluates declared capabilities against rules from four sources (builtin, user `~/.{app_name}/settings.json`, project `./.{app_name}/settings.json`, in-memory session). ADK + LangGraph gate tool calls via `workflow/adk/permission_plugin.py::PermissionPlugin` and `workflow/langgraph/permission_wrap.py::wrap_tool_for_permission`.
 - **Service registry**: Tools access services and shared state via `get_service(key)` from `workflow.service_registry`. A single ContextVar holds a `dict[str, Any]` set by the workflow manager during processing. Complex services (KBManager, SandboxManager, MemoryStore) are lazily created; simple state (plan string, task list) lives directly in the registry dict.
-- **Manager detection**: `BaseWorkflowManager._detect_required_managers()` scans each agent's tool names against the `_TOOL_SERVICE_MAP` (name → service key, in `base_manager.py`); `_ensure_managers_initialized()` then lazily instantiates only the services actually needed (KBManager, SandboxManager, MemoryStore, …). Adding a new service-backed tool means adding its name → service entry to `_TOOL_SERVICE_MAP`. (There is no `@requires` decorator.)
+- **Manager detection**: tools declare their own service needs — `@register_tool(..., requires="kb_manager")` (or a tuple for several) — and the key is validated against `service_registry.KNOWN_SERVICE_KEYS` at registration (only *constructible* services are declarable; `user_kb_manager` is created together with `kb_manager`). `BaseWorkflowManager._detect_required_managers()` reads that metadata off the registry for each agent's tools (by registry identity, so `register(func, name=...)`'s original callable still declares its services); `_build_services()` then lazily constructs only the services actually needed, into a local dict that is released in full if a later constructor raises (nothing is published, so nothing else could close it). A downstream tool may request any framework-provided service without editing the framework; there is no mechanism for registering new service *types*, and no central name→service map.
+- **Canonical tool names + permission identity**: `ToolDefinition.name` is the single identity. `register_tool(name="public_name")` wraps the callable so `func.__name__` is the registered name (backends derive the model-visible name from the callable). Permission gating **and tool assembly** resolve through a **registry-owned identity binding**: each `ToolRegistry` owns a `id(obj) → (weakref, definition)` map (`registry.bind_identity()`/`identify()`; the module-level `tools.registry.bind_tool_identity()`/`identify_tool()` answer for `get_registry()`). Every hit is confirmed with `is` against the weak reference, so neither a name, a class name, nor a forged `__eq__`/`__hash__` can stand in for it, and a recycled address inherits nothing. Identity is **per registry** — a tool registered into an application's own `ToolRegistry` is not one of the framework's, so it stays untouched during assembly and is denied at permission time — which is also what lets a short-lived registry, its definitions and their closures be garbage collected. Everything the framework issues is bound at its construction site: registered callables, factory service-bound variants, renamed wrappers, and the native ADK tool objects the framework builds (skill tools, in `tools/skills/toolset.py`). `workflow/adk/permission_plugin.py` unwraps `.func` only for the exact ADK types whose contract is to call it (`FunctionTool`, `LongRunningFunctionTool`), and gates genuine `McpTool` instances (`isinstance`, since ADK creates them on connect) under a synthetic `mcp` capability. Anything unbound is denied — and, in assembly (service detection, service-tool substitution, canonicalization, long-running wrapping), left exactly as the application supplied it: a plain callable named `kb_search` is not the framework's tool and must not be given its services, its service-bound variant, or its long-running contract. Substitution additionally requires the service variant to *be* that same definition (`identify_tool(variant) is definition`); factories bind each closure to the exact module-level tool it re-binds, so a tool an application has taken over keeps its own implementation.
+- **One name, one tool**: `ToolRegistry.register()` raises on any name that is already registered — matching capabilities are *not* grounds for sharing one, since they say nothing about the docstring or the model-visible schema. Sharing is declared, never inferred: `declare_tool(name, ...)` (exported from `agentic_cli.tools`) declares a tool that has **no backend-neutral implementation** (`ToolDefinition.func is None`), and each backend registers its own with `register_tool(..., variant_of=name)` — same identity and permission contract, its own signature and docstring. Re-declaring the same contract is idempotent; changing its description or capabilities raises. `definition.variants` is ordered by defining module, so it never depends on import order, and assembly substitutes a variant's canonical callable via `registry.canonical_for()` (never `None`). A `replace=True` retires the previous definition's callables, and retired backend variants are excluded from `include_state_tools` injection, so the model never sees two tools with one name. That is how the ADK and LangGraph `save_plan`/`get_plan`/`save_tasks`/`get_tasks` coexist (declared in `tools/_core/state_tools.py`); previously they contested the name and import order decided the winner. A bare-name reference to a declared-only tool raises "ambiguous" rather than guessing a backend. `replace=True` takes a name over deliberately and **retires the old definition's identities**, so its callables resolve to nothing (denied, and left alone by assembly) rather than inheriting the replacement's capabilities.
+- **Turn/lifecycle concurrency**: a manager runs one turn at a time — `process()`/`resume_with_job_result()` enter through `_turn_admission()` (which holds `_turn_lock`), and `initialize_services`/`reinitialize`/`cleanup` hold `_lifecycle_lock` **and** `_turn_lock`. Lock order is lifecycle → turn; a turn initializes *before* taking the turn lock, which is what keeps the two from deadlocking — and because of that a cleanup can land in between, so admission re-checks `_backend_ready()` while holding the turn lock and reinitializes once (or fails cleanly) rather than running against released resources. Initialization is transactional: services are built on a worker thread into a *local* dict and published only while the attempt still owns init (a cancelled attempt releases what the thread went on to build), and a failed attempt rolls back. A failed in-place reinitialization leaves the manager uninitialized; the controller reports `FAILED` and refuses to hand it out, but **keeps** it so a retry can revive it with its preserved (possibly in-memory) sessions intact. `WorkflowController` serializes init/reinitialize/swap/close on its own lifecycle lock, and a background init that finishes after `close()` releases its manager instead of publishing it.
+- **HITL callback is context-local**: `set_input_callback()` stores into a per-manager `ContextVar`, so a second consumer installing its callback cannot capture a running turn's prompt, and one consumer's `clear_input_callback()` cannot unregister another's. `MessageProcessor` cancels and awaits its consumer task **before** clearing the callback, so no tool is left asking a question nobody owns.
+- **No harness-level turn replay**: ADK persists a turn's input during invocation setup, so the CLI never re-invokes an event source. Retries belong to the provider client (`HttpRetryOptions`). `MessageProcessor` returns a typed `TurnResult`.
+- **Session identity**: durable sessions are addressed by `SessionRef(app_name, user_id, session_id)` (`workflow/sessions.py`). Every session hook (`session_exists`/`list_sessions`/`delete_session`/`recent_messages`/`load_session`) takes an optional `user_id`, defaulting to `settings.default_user` only when the caller omits it. Backends without a session store leave `supports_sessions` False, and the base hooks raise `NotImplementedError` rather than answering with a misleading `False`/`[]`.
+- **Active turn**: the in-flight `(user, session)` is a `ContextVar` (`workflow/sessions.py::get_active_turn`), set with a token by `_workflow_context()` — concurrent turns on one manager stay isolated and nesting restores the outer turn.
+- **Resource ownership**: `cleanup()` is idempotent and awaits an async `close()` on owned resources (`BaseWorkflowManager._aclose_owned`); `WorkflowController.close()` is the single shutdown path (cancel init → shut executor → clean manager).
 - **Atomic writes**: Use `atomic_write_json`/`atomic_write_text` from `file_utils.py` for file persistence.
 
 ### Console Output

README.md

@@ -281,6 +281,11 @@ dynamic_agent = AgentConfig(
     tools=[tool_a, tool_b],
 )
 
+# A prompt factory may also take the manager's settings explicitly. Either way
+# it is evaluated under the manager's settings, not the global singleton.
+def get_scoped_prompt(settings):
+    return f"You are {settings.app_name}."
+
 # Coordinator with sub-agents
 coordinator = AgentConfig(
     name="coordinator",
@@ -310,13 +315,62 @@ configs = [coordinator, researcher, analyst]
 
 | Field | Type | Description |
 |-------|------|-------------|
-| `name` | str | Unique identifier |
-| `prompt` | str \| Callable | System instruction |
+| `name` | str | Unique identifier (must be unique across the config list) |
+| `prompt` | str \| Callable | System instruction; a callable may take no arguments or a single `settings` argument |
 | `tools` | list[Callable] | Available tool functions |
 | `sub_agents` | list[str] | Names of agents this one can delegate to |
-| `description` | str | Short description for routing |
+| `description` | str | Short description for routing (defaults to `""`) |
 | `model` | str \| None | Model override (defaults to manager's model) |
 
+The agent graph is validated before anything is allocated — ahead of model
+discovery and service creation. Duplicate names, unknown `sub_agents`
+references, self-references, delegation cycles, a sub-agent shared by two
+parents, and **more than one root** raise `AgentGraphError` naming the
+offending agents. Exactly one agent may be unreferenced: the runner starts from
+a single root, so agents under any other root would never run. Agents are built
+in dependency order, so declaration order does not matter. A per-agent `model`
+override is validated against the configured provider credentials at startup,
+alongside `default_model`.
+
+A callable `prompt` may take **no arguments** (including all-defaulted ones) or
+**exactly one argument**, which receives the manager's settings. Other
+signatures, `async def` factories, and non-string results raise
+`AgentGraphError` naming the agent.
+
+### Turn and lifecycle contracts
+
+A workflow manager runs **one turn at a time**: `process()` and
+`resume_with_job_result()` serialize on a turn lock, and
+`initialize_services()`/`reinitialize()`/`cleanup()` take it too, so the backend
+is never torn down mid-stream. Admission also re-verifies the backend is live
+after acquiring the lock, so a turn queued behind a cleanup reinitializes (or
+fails cleanly) instead of running against released resources. Run separate
+managers for genuine parallelism.
+
+`MessageProcessor.process()` returns a `TurnResult` (`TurnStatus.COMPLETED` /
+`CANCELLED` / `FAILED` / `UNAVAILABLE`); a turn is never replayed by the
+harness, so a surfaced rate limit fails it explicitly. An `EventType.ERROR`
+event is rendered as it arrives — `recoverable=True` is a warning and the
+stream still decides the outcome, anything else makes the turn `FAILED`
+(`delivered` False). Cancelling the caller cancels and awaits the event
+consumer before the turn is torn down.
+
+The HITL input callback is **context-local**: `set_input_callback()` binds it
+for the calling context (and any task started from it), so two consumers of one
+manager cannot capture each other's prompts.
+
+`WorkflowController` exposes a derived `WorkflowState`
+(`uninitialized`/`initializing`/`ready`/`failed`/`closed`) that never reports
+`ready` over an uninitialized manager, and `controller.workflow` raises unless
+the state is `ready`. Lifecycle transitions (init, reinitialize, orchestrator
+swap, close) are serialized, so nothing is published after `close()`. A failed
+in-place reinitialization keeps the manager — it still owns the session service
+it preserved — and the next initialization revives it rather than discarding
+the conversation.
+
+`SessionRef(app_name, user_id, session_id)` is the conversation identity used by
+every session API. All of these are importable from `agentic_cli`.
+
 ## Tools
 
 ### Creating Tools

examples/jobs_demo.py

@@ -61,6 +61,9 @@
 @register_tool(
     category=ToolCategory.EXECUTION,
     capabilities=[Capability("longrunning.run_shell_job")],
+    # Declares its own service need, so the manager creates the JobManager for
+    # an app that ships only this starter (no framework job_* tools required).
+    requires="job_manager",
     long_running=True,
     description="Run a shell command as a detached background job; returns a job_id immediately.",
 )

src/agentic_cli/__init__.py

@@ -25,9 +25,12 @@
 from agentic_cli.cli.app import BaseCLIApp
 from agentic_cli.workflow.factory import create_workflow_manager_from_settings
 from agentic_cli.cli.commands import Command, CommandRegistry
-from agentic_cli.workflow.config import AgentConfig
+from agentic_cli.cli.message_processor import TurnResult, TurnStatus
+from agentic_cli.cli.workflow_controller import WorkflowState
+from agentic_cli.workflow.config import AgentConfig, AgentGraphError
 from agentic_cli.workflow.model_settings import ModelSettings, ThinkingSettings
 from agentic_cli.workflow.events import WorkflowEvent, EventType
+from agentic_cli.workflow.sessions import SessionRef
 from agentic_cli.config import (
     BaseSettings,
     SettingsContext,
@@ -74,10 +77,16 @@ def __getattr__(name: str):
     "GoogleADKWorkflowManager",  # lazy (Google ADK)
     "LangGraphWorkflowManager",  # lazy (requires langgraph extra)
     "AgentConfig",
+    "AgentGraphError",
     "ModelSettings",
     "ThinkingSettings",
     "WorkflowEvent",
     "EventType",
+    # Lifecycle / turn contracts
+    "SessionRef",
+    "TurnResult",
+    "TurnStatus",
+    "WorkflowState",
     # Settings
     "BaseSettings",
     "SettingsContext",

src/agentic_cli/cli/app.py

@@ -476,29 +476,88 @@ async def resume_finished_jobs(self) -> int:
         """Resume the agent for each finished, resume-flagged background job.
 
         Each resume is a serialized turn (via the turn lock) so it never
-        overlaps a user turn or another resume. Returns the number resumed.
-        Used at turn boundaries (auto, gated) and by the /resume command
-        (explicit, ungated).
+        overlaps a user turn or another resume. Used at turn boundaries (auto,
+        gated) and by the /resume command (explicit, ungated).
+
+        Delivery follows the job's resume lifecycle: the job is *claimed*
+        (pending → resuming) before the turn runs, so a crash cannot silently
+        re-deliver it, and only recorded delivered once the turn actually
+        completed — a failed or cancelled resume is recorded as such instead of
+        being dropped.
+
+        Returns:
+            The number of jobs a resume turn was run for (unchanged meaning:
+            "how many were picked up"). Whether each one was delivered is
+            recorded on the job and reported by ``/jobs``; a failed delivery is
+            surfaced to the user by the turn itself.
         """
         if not self._workflow_controller.is_ready:
             return 0
         jm = getattr(self._workflow_controller.workflow, "job_manager", None)
         if jm is None:
             return 0
 
-        records = jm.awaiting_resume()
-        for record in records:
-            # Mark before running so a crash mid-resume can't double-fire.
-            jm.mark_resumed(record.job_id)
+        attempted = 0
+        for record in jm.awaiting_resume():
+            # Claim first: durable, so an interrupted delivery is recoverable
+            # and two coordinators can't both deliver the same result.
+            if not jm.begin_resume(record.job_id):
+                continue
+            attempted += 1
+            await self._deliver_resume(jm, record)
+        return attempted
+
+    async def _deliver_resume(self, jm, record) -> bool:
+        """Run one claimed job's resume turn and close out its transition.
+
+        Every exit path closes the claim, so no record can be left ``RESUMING``
+        in this process: a normal outcome records delivered/failed, an
+        exception records failed with the reason, and a cancellation records
+        failed before re-raising (cancellation still propagates).
+
+        Args:
+            jm: The JobManager holding the claim.
+            record: The claimed job record.
+
+        Returns:
+            True if the result was delivered to the agent.
+        """
+        try:
             async with self._turn_lock:
-                await self._message_processor.process_resume(
+                result = await self._message_processor.process_resume(
                     record=record,
                     workflow_controller=self._workflow_controller,
                     ui=self.session,
                     settings=self._settings,
                     usage_tracker=self._usage_tracker,
                 )
-        return len(records)
+        except asyncio.CancelledError:
+            jm.complete_resume(
+                record.job_id, delivered=False, error="resume cancelled"
+            )
+            logger.info("job_resume_cancelled", job_id=record.job_id)
+            raise
+        except Exception as exc:  # noqa: BLE001 - the claim must always close
+            jm.complete_resume(record.job_id, delivered=False, error=str(exc))
+            logger.warning(
+                "job_resume_raised", job_id=record.job_id, error=str(exc)
+            )
+            self.session.add_error(
+                f"Background job '{record.name}' could not be resumed: {exc}"
+            )
+            return False
+
+        jm.complete_resume(
+            record.job_id, delivered=result.delivered, error=result.error
+        )
+        if not result.delivered:
+            logger.info(
+                "job_resume_not_delivered",
+                job_id=record.job_id,
+                status=result.status.value,
+                error=result.error,
+            )
+        return result.delivered
 
     async def _adopt_session_on_startup(self) -> None:
         """Adopt this run's session id so the manager targets it from turn one.
@@ -571,8 +630,11 @@ async def handle_input(text: str) -> None:
                 # Run the session - user sees prompt immediately!
                 await self.session.run_async()
 
-        # Extract session facts into memory on exit (if enabled)
-        await self._extract_session_facts_on_exit()
+            # Extract session facts into memory (if enabled) while the workflow
+            # is still alive: leaving this context closes the manager, and fact
+            # extraction needs the live session store and an LLM call.
+            await self._extract_session_facts_on_exit()
+
         # No save-on-exit: durable session stores persist continuously per turn.
 
         logger.info("app_ending")