tiphys.kernel

21class TiphysGuard:
22    """L1 Guard: validates, budgets, and gates tool execution via Castor.
23
24    Pipeline:
25        1. Gate.validate(tool_name, arguments) → validated args or error
26        2. Check destructive/requires_hitl → call on_hitl callback
27        3. cap_mgr.deduct(capabilities, resource, cost)
28        4. Gate.execute(tool_name, validated_args)
29        5. Return result
30    """
31
32    def __init__(
33        self,
34        gate: SyscallGate,
35        capability_manager: CapabilityManager,
36    ) -> None:
37        self._gate = gate
38        self._cap_mgr = capability_manager
39
40    @property
41    def gate(self) -> SyscallGate:
42        """Access the underlying SyscallGate."""
43        return self._gate
44
45    @property
46    def capability_manager(self) -> CapabilityManager:
47        """Access the underlying CapabilityManager."""
48        return self._cap_mgr
49
50    async def execute_tool(
51        self,
52        tool_name: str,
53        arguments: dict[str, Any],
54        capabilities: dict[str, Capability] | None = None,
55        on_hitl: Callable[[str, dict[str, Any]], Awaitable[bool]] | None = None,
56    ) -> Any:
57        """Execute a tool through the guard pipeline.
58
59        Args:
60            tool_name: Name of the tool to execute.
61            arguments: Raw arguments from the LLM.
62            capabilities: Budget capabilities for this session (None = no budget).
63            on_hitl: Async callback for HITL approval. Receives (tool_name, args),
64                     returns True to proceed, False to reject.
65
66        Returns:
67            Tool execution result.
68
69        Raises:
70            ToolExecutionError: If validation fails or HITL is rejected.
71            CapabilityExhaustedError: If budget is exhausted.
72        """
73        meta = self._gate.get_tool_meta(tool_name)
74
75        # 1. Validate arguments
76        try:
77            validated = self._gate.validate(tool_name, arguments)
78        except ValidationError as e:
79            response = self._gate.format_validation_error(tool_name, e)
80            raise ToolExecutionError(tool_name, response.feedback_message or str(e)) from e
81
82        # 2. HITL gate
83        if (meta.destructive or meta.requires_hitl) and on_hitl is not None:
84            approved = await on_hitl(tool_name, validated)
85            if not approved:
86                raise ToolExecutionError(tool_name, "Action rejected by user")
87
88        # 3. Budget deduction (Castor V2.1: deduct() no-ops for untracked resources)
89        if capabilities:
90            self._cap_mgr.deduct(capabilities, meta.consumes, meta.cost_per_use)
91
92        # 4. Execute
93        result: Any = await self._gate.execute(tool_name, validated)
94        return result

21class HITLBridge:
22    """Bridges Castor HITL decisions to Tiphys's EventBus permission flow.
23
24    When a tool requires HITL approval, the bridge emits a
25    PermissionRequestEvent and waits for the corresponding
26    PermissionResponseEvent from the user.
27    """
28
29    def __init__(self, event_bus: EventBus, timeout: float = 300.0) -> None:
30        self._event_bus = event_bus
31        self._timeout = timeout
32        self._pending: dict[str, asyncio.Future[PermissionResponseEvent]] = {}
33        self._event_bus.subscribe("permission_response", self._on_response)
34
35    async def request_approval(
36        self,
37        tool_name: str,
38        arguments: dict[str, Any],
39        agent_id: str,
40    ) -> bool:
41        """Request HITL approval via EventBus.
42
43        Emits PermissionRequestEvent and waits for PermissionResponseEvent.
44
45        Args:
46            tool_name: The tool requesting approval.
47            arguments: The tool's validated arguments.
48            agent_id: The agent making the request.
49
50        Returns:
51            True if approved, False if rejected or timed out.
52        """
53        event = PermissionRequestEvent(
54            action=f"execute:{tool_name}",
55            details=str(arguments),
56            agent_id=agent_id,
57        )
58
59        loop = asyncio.get_running_loop()
60        future: asyncio.Future[PermissionResponseEvent] = loop.create_future()
61        self._pending[event.request_id] = future
62
63        await self._event_bus.broadcast(event)
64
65        try:
66            response = await asyncio.wait_for(future, timeout=self._timeout)
67            return response.approved
68        except TimeoutError:
69            return False
70        finally:
71            self._pending.pop(event.request_id, None)
72
73    async def _on_response(self, event: BaseEvent) -> None:
74        """Handle incoming PermissionResponseEvent."""
75        if not isinstance(event, PermissionResponseEvent):
76            return
77
78        future = self._pending.get(event.request_id)
79        if future is not None and not future.done():
80            future.set_result(event)

14class SessionBudgetManager:
15    """Creates and tracks Castor capability budgets per session.
16
17    Budget defaults come from BudgetConfig, with optional
18    per-channel overrides.
19    """
20
21    def __init__(
22        self,
23        budget_config: BudgetConfig,
24        capability_manager: CapabilityManager,
25    ) -> None:
26        self._config = budget_config
27        self._cap_mgr = capability_manager
28        self._sessions: dict[str, dict[str, Capability]] = {}
29
30    def get_or_create(
31        self,
32        session_key: str,
33        channel: str | None = None,
34    ) -> dict[str, Capability]:
35        """Get existing budget or create from config defaults.
36
37        Args:
38            session_key: Unique session identifier.
39            channel: Optional channel name for per-channel overrides.
40
41        Returns:
42            Capability budget dict for the session.
43        """
44        if session_key in self._sessions:
45            return self._sessions[session_key]
46
47        budgets = self._config.default_budgets.copy()
48        if channel and channel in self._config.per_channel_overrides:
49            budgets.update(self._config.per_channel_overrides[channel])
50
51        caps = self._cap_mgr.create_capabilities(budgets)
52        self._sessions[session_key] = caps
53        return caps
54
55    def get(self, session_key: str) -> dict[str, Capability] | None:
56        """Get budget for an existing session, or None."""
57        return self._sessions.get(session_key)
58
59    def reset(self, session_key: str) -> None:
60        """Clear budget for a session."""
61        self._sessions.pop(session_key, None)

18class ToolCostConfig(BaseModel):
19    """Per-tool cost configuration for Castor budget tracking."""
20
21    consumes: str = Field(default="api", description="Resource type this tool consumes")
22    cost_per_use: float = Field(default=1.0, ge=0.0, description="Cost per invocation")

25def adapt_tool(
26    tool: ToolDefinition,
27    cost_config: ToolCostConfig | None = None,
28) -> ToolMetadata:
29    """Convert a Tiphys ToolDefinition to a Castor ToolMetadata."""
30    cfg = cost_config or ToolCostConfig()
31
32    meta = ToolMetadata.from_function(
33        tool.handler,
34        consumes=cfg.consumes,
35        cost_per_use=cfg.cost_per_use,
36        requires_hitl=tool.requires_permission,
37        destructive=tool.requires_permission,
38    )
39    meta.tool_name = tool.name
40    return meta

43def register_chela_tools(
44    chela_registry: TiphysToolRegistry,
45    castor_registry: CastorToolRegistry,
46    cost_config: dict[str, ToolCostConfig] | None = None,
47) -> None:
48    """Register all Tiphys tools into a Castor ToolRegistry.
49
50    Args:
51        chela_registry: Source registry with Tiphys tool definitions.
52        castor_registry: Target Castor registry to populate.
53        cost_config: Optional per-tool cost overrides keyed by tool name.
54    """
55    costs = cost_config or {}
56
57    for name, tool in chela_registry.get_all().items():
58        meta = adapt_tool(tool, costs.get(name))
59        castor_registry.register(meta)

 40class CastorAgentRunner:
 41    """Agent runner backed by Castor's checkpoint/replay kernel.
 42
 43    Instead of the standard AgentRunner's direct LLM/tool calls,
 44    all operations are routed through Castor's SyscallProxy. This gives:
 45
 46    - Deterministic replay from checkpoint on resume
 47    - Crash recovery via CheckpointStore
 48    - HITL suspend/resume with checkpoint persistence
 49    - Budget enforcement on both LLM and tool calls
 50    """
 51
 52    def __init__(
 53        self,
 54        context: AppContext,
 55        llm_client: LLMClient | None = None,
 56    ) -> None:
 57        self._context = context
 58        self._config = context.config
 59        self._event_bus: EventBus = context.event_bus
 60        self._chela_tools: TiphysToolRegistry = context.tool_registry
 61        self._llm = llm_client or LLMClient.from_config(self._config)
 62        self._session_store = get_session_store()
 63
 64        # Build Castor tool registry from Tiphys tools
 65        self._castor_registry = CastorToolRegistry()
 66        register_chela_tools(self._chela_tools, self._castor_registry)
 67
 68        # Create LLM syscall wrapper
 69        self._llm_syscall: LLMSyscall = create_llm_syscall(self._llm, self._castor_registry)
 70
 71        # Use the context's kernel if available, otherwise build one
 72        if context.castor_kernel is not None:
 73            self._kernel = context.castor_kernel
 74        else:
 75            gate = SyscallGate(self._castor_registry)
 76            cap_mgr = CapabilityManager()
 77            default_budgets = (
 78                self._config.budget.default_budgets if self._config.budget.enabled else None
 79            )
 80            self._kernel = Castor(
 81                gate=gate,
 82                capability_manager=cap_mgr,
 83                store=context.checkpoint_store,
 84                default_budgets=default_budgets,
 85            )
 86
 87    async def run(
 88        self,
 89        session_key: str,
 90        user_message: str,
 91        *,
 92        max_steps: int = 10,
 93        budgets: dict[str, float] | None = None,
 94        checkpoint: AgentCheckpoint | None = None,
 95        pid: str | None = None,
 96    ) -> str:
 97        """Run the agent to completion using Castor's kernel.
 98
 99        Args:
100            session_key: Session identifier for history persistence.
101            user_message: The user's message.
102            max_steps: Maximum ReAct loop iterations.
103            budgets: Resource budgets (e.g. ``{"api": 50.0}``).
104            checkpoint: Existing checkpoint to resume from.
105            pid: Custom process ID.
106
107        Returns:
108            The agent's final text response.
109        """
110        # Load session for message history
111        session = self._session_store.get_or_create(session_key, self._config.agent_id)
112        session.add_message(role="user", content=user_message)
113
114        # Build the context messages for the LLM
115        tools_list = list(self._chela_tools.get_all().values())
116        system_prompt = build_system_prompt(
117            mode="full",
118            agent_name=self._config.agent_name,
119            tools=tools_list,
120        )
121
122        tool_schemas = [
123            {
124                "type": "function",
125                "function": {
126                    "name": t.name,
127                    "description": t.description,
128                    "parameters": t.args_schema.model_json_schema(),
129                },
130            }
131            for t in tools_list
132        ]
133
134        messages = self._build_llm_messages(session, system_prompt)
135        event_bus = self._event_bus
136        agent_id = self._config.agent_id
137        llm_syscall = self._llm_syscall
138
139        async def agent_fn(proxy: SyscallProxy) -> str:
140            """The agent coroutine executed inside Castor's kernel."""
141            current_messages = list(messages)
142            steps = 0
143
144            while steps < max_steps:
145                # LLM call via syscall (replay-safe)
146                llm_result = await llm_syscall.infer(
147                    proxy,
148                    messages=[m.model_dump(exclude_none=True) for m in current_messages],
149                    tools=tool_schemas if tool_schemas else None,
150                )
151
152                content = llm_result.get("content")
153                tool_calls = llm_result.get("tool_calls", [])
154
155                # Emit thinking/text event
156                if content:
157                    await event_bus.broadcast(
158                        TextEvent(
159                            content=content,
160                            agent_id=agent_id,
161                            is_final=not tool_calls,
162                        )
163                    )
164
165                if not tool_calls:
166                    return content or ""
167
168                # Build assistant message with tool calls
169                assistant_msg = Message(
170                    role="assistant",
171                    content=content,
172                    tool_calls=[
173                        {
174                            "id": tc["id"],
175                            "type": "function",
176                            "function": {
177                                "name": tc["name"],
178                                "arguments": str(tc["arguments"]),
179                            },
180                        }
181                        for tc in tool_calls
182                    ],
183                )
184                current_messages.append(assistant_msg)
185
186                # Execute each tool call via syscall
187                for tc in tool_calls:
188                    tool_name = tc["name"]
189                    tool_args = tc["arguments"]
190                    tool_call_id = tc["id"]
191
192                    await event_bus.broadcast(
193                        ToolCallEvent(
194                            tool_name=tool_name,
195                            arguments=tool_args,
196                            tool_call_id=tool_call_id,
197                            agent_id=agent_id,
198                        )
199                    )
200
201                    # Tool execution via syscall (replay-safe, budget-tracked)
202                    result = await proxy.syscall(tool_name, tool_args)
203
204                    result_str = str(result)
205                    await event_bus.broadcast(
206                        ToolResultEvent(
207                            tool_call_id=tool_call_id,
208                            tool_name=tool_name,
209                            result=result_str,
210                            success=True,
211                            agent_id=agent_id,
212                        )
213                    )
214
215                    current_messages.append(
216                        Message(
217                            role="tool",
218                            content=result_str,
219                            name=tool_name,
220                            tool_call_id=tool_call_id,
221                        )
222                    )
223
224                steps += 1
225
226            return "Reached maximum steps."
227
228        # Run through Castor kernel
229        result_cp = await self._kernel.run(
230            agent_fn,
231            budgets=budgets,
232            checkpoint=checkpoint,
233            pid=pid or f"chela-{uuid.uuid4().hex[:8]}",
234        )
235
236        # Handle suspension for HITL
237        if result_cp.status == "SUSPENDED_FOR_HITL":
238            await self._handle_hitl_suspension(result_cp)
239            return f"Action requires approval: {result_cp.pending_tool}({result_cp.pending_args})"
240
241        # Persist final response to session
242        final_text = str(result_cp.result) if result_cp.result else ""
243        session.add_message(role="assistant", content=final_text)
244        self._session_store.save(session)
245
246        # Save completed checkpoint
247        if self._kernel.store is not None:
248            await self._kernel.save(result_cp)
249
250        return final_text
251
252    async def resume(
253        self,
254        checkpoint: AgentCheckpoint,
255        session_key: str,
256        *,
257        max_steps: int = 10,
258    ) -> str:
259        """Resume a suspended agent from a checkpoint.
260
261        Args:
262            checkpoint: The checkpoint to resume (must be RUNNING after
263                        approve/reject/modify has been applied).
264            session_key: Session key for history persistence.
265            max_steps: Maximum remaining ReAct iterations.
266
267        Returns:
268            The agent's final text response.
269        """
270        return await self.run(
271            session_key=session_key,
272            user_message="",
273            max_steps=max_steps,
274            checkpoint=checkpoint,
275        )
276
277    async def _handle_hitl_suspension(self, checkpoint: AgentCheckpoint) -> None:
278        """Delegate HITL suspension to the checkpoint bridge."""
279        bridge = self._context.checkpoint_hitl_bridge
280        if bridge is not None:
281            await bridge.handle_suspension(checkpoint, self._config.agent_id)
282        else:
283            # Fallback: emit event directly (no bridge available)
284            tool_name = checkpoint.pending_tool or "unknown"
285            args = checkpoint.pending_args or {}
286            event = PermissionRequestEvent(
287                action=f"execute:{tool_name}",
288                details=str(args),
289                agent_id=self._config.agent_id,
290            )
291            await self._event_bus.broadcast(event)
292            # Persist checkpoint for later resume
293            if self._kernel.store is not None:
294                await self._kernel.save(checkpoint)
295
296    def _build_llm_messages(
297        self,
298        session: Any,
299        system_prompt: str,
300    ) -> list[Message]:
301        """Build the message list for the LLM from session history."""
302        messages: list[Message] = [Message(role="system", content=system_prompt)]
303
304        for msg in session.messages:
305            if msg.role != "system":
306                messages.append(
307                    Message(
308                        role=msg.role,
309                        content=msg.content,
310                        name=msg.name,
311                        tool_call_id=msg.tool_call_id,
312                        tool_calls=msg.tool_calls,
313                    )
314                )
315
316        return messages

 83class CheckpointHITLBridge:
 84    """L2 checkpoint-aware HITL bridge for suspend/resume cycles.
 85
 86    Manages the full lifecycle: suspend → persist checkpoint → emit event
 87    → await user decision → apply decision → resume agent.
 88    """
 89
 90    def __init__(
 91        self,
 92        event_bus: EventBus,
 93        kernel: Castor,
 94        timeout: float = 300.0,
 95    ) -> None:
 96        self._event_bus = event_bus
 97        self._kernel = kernel
 98        self._timeout = timeout
 99        self._pending: dict[str, asyncio.Future[PermissionResponseEvent]] = {}
100        self._event_bus.subscribe("permission_response", self._on_response)
101
102    async def handle_suspension(
103        self,
104        checkpoint: AgentCheckpoint,
105        agent_id: str,
106    ) -> None:
107        """Handle a checkpoint suspended for HITL.
108
109        Persists the checkpoint and emits a PermissionRequestEvent
110        so the user can approve/reject/modify.
111
112        Args:
113            checkpoint: Suspended checkpoint with pending_hitl set.
114            agent_id: Agent identifier for the event.
115        """
116        # Persist checkpoint so it survives gateway restarts
117        if self._kernel.store is not None:
118            await self._kernel.save(checkpoint)
119
120        tool_name = checkpoint.pending_tool or "unknown"
121        args = checkpoint.pending_args or {}
122
123        event = PermissionRequestEvent(
124            action=f"execute:{tool_name}",
125            details=str(args),
126            agent_id=agent_id,
127        )
128        await self._event_bus.broadcast(event)
129
130    async def await_decision(
131        self,
132        checkpoint: AgentCheckpoint,
133        agent_id: str,
134    ) -> tuple[str, str | None]:
135        """Await user decision on a suspended checkpoint.
136
137        Emits PermissionRequestEvent and waits for response.
138
139        Args:
140            checkpoint: Suspended checkpoint.
141            agent_id: Agent identifier.
142
143        Returns:
144            Tuple of (decision, feedback) where decision is
145            "approve", "reject", or "modify".
146        """
147        tool_name = checkpoint.pending_tool or "unknown"
148        args = checkpoint.pending_args or {}
149
150        event = PermissionRequestEvent(
151            action=f"execute:{tool_name}",
152            details=str(args),
153            agent_id=agent_id,
154        )
155
156        loop = asyncio.get_running_loop()
157        future: asyncio.Future[PermissionResponseEvent] = loop.create_future()
158        self._pending[event.request_id] = future
159
160        await self._event_bus.broadcast(event)
161
162        try:
163            response = await asyncio.wait_for(future, timeout=self._timeout)
164            if response.approved:
165                if response.feedback:
166                    return ("modify", response.feedback)
167                return ("approve", None)
168            return ("reject", response.feedback)
169        except TimeoutError:
170            return ("reject", "Timed out waiting for approval")
171        finally:
172            self._pending.pop(event.request_id, None)
173
174    async def apply_decision(
175        self,
176        checkpoint: AgentCheckpoint,
177        decision: str,
178        feedback: str | None = None,
179    ) -> None:
180        """Apply a HITL decision to the checkpoint.
181
182        Args:
183            checkpoint: The suspended checkpoint.
184            decision: One of "approve", "reject", "modify".
185            feedback: Optional feedback text for reject/modify.
186        """
187        if decision == "approve":
188            await self._kernel.approve(checkpoint)
189        elif decision == "reject":
190            self._kernel.reject(checkpoint, feedback or "")
191        elif decision == "modify":
192            self._kernel.modify(checkpoint, feedback or "")
193
194        # Update persisted checkpoint
195        if self._kernel.store is not None:
196            await self._kernel.save(checkpoint)
197
198    async def _on_response(self, event: BaseEvent) -> None:
199        """Handle incoming PermissionResponseEvent."""
200        if not isinstance(event, PermissionResponseEvent):
201            return
202
203        future = self._pending.get(event.request_id)
204        if future is not None and not future.done():
205            future.set_result(event)