"""The `sdlc_*` → Node bridge path: every SDLC tool shells out to `npx rstack-bridge` (bin/rstack-bridge.ts), which reuses the existing TypeScript adapter and harness verbatim — no SDLC logic is reimplemented in Python. owner: RStack developed by Richardson Gunde; Tau adapter contributed by Jeomon """ from __future__ import annotations import asyncio import json import os from typing import Optional from tau.tool.types import Tool, ToolContext, ToolExecutionMode, ToolInvocation, ToolKind, ToolResult from .config import _timeout_from_env from .proc import _communicate_or_kill, _own_process_group_kwargs from .resolve import _resolve_bin_argv # G-06 (#552): named timeout constants. Long-running tools get a bound that # OUTLIVES the inner timeout mechanism their bridge call hosts, by a fixed # margin, so the inner (better-scoped) mechanism always fires first. _MINUTE_MS = 60_000 # Default for control-plane sdlc_* calls; may absorb a cold `npx` fetch. _CONTROL_PLANE_TIMEOUT_MS = 60_000 # Margin added on top of each inner backstop below. _INNER_BACKSTOP_MARGIN_MS = _MINUTE_MS # Pi's delegate backstop: 30 minutes (DEFAULT_DELEGATE_TIMEOUT_MS, # src/integrations/pi/rstack-sdlc.ts). The old uniform 60s bound killed every # delegation driven through this adapter at 1/30th of it. _PI_DELEGATE_BACKSTOP_MS = 30 * _MINUTE_MS # Sandbox test-execution cap: 10 minutes (MAX_TIMEOUT_MS, # src/core/harness/sandbox.js). _SANDBOX_EXECUTION_CAP_MS = 10 * _MINUTE_MS # tool → (dedicated override env var, default ms). _LONG_RUNNING_TOOL_TIMEOUTS = { "sdlc_delegate": ("RSTACK_DELEGATE_TIMEOUT_MS", _PI_DELEGATE_BACKSTOP_MS + _INNER_BACKSTOP_MARGIN_MS), "sdlc_validate": ("RSTACK_VALIDATE_TIMEOUT_MS", _SANDBOX_EXECUTION_CAP_MS + _INNER_BACKSTOP_MARGIN_MS), } def _bridge_timeout_s(tool: str = "") -> float: """Hard bound on a single bridge invocation — the adapter's PRIMARY tool-call path, invoked on every single sdlc_* call (#487). Per-tool policy (G-06, #552): control-plane tools keep the 60s default (bridge calls may cold-start `npx`, which the shorter guard/gate bounds elsewhere deliberately don't need to absorb), overridable via RSTACK_BRIDGE_TIMEOUT_MS. Long-running tools get a bound that OUTLIVES their inner backstop (see _LONG_RUNNING_TOOL_TIMEOUTS), each with its own dedicated override env — the global RSTACK_BRIDGE_TIMEOUT_MS deliberately does NOT re-cap them, since a uniform cap silently killing long delegations was exactly the G-06 defect.""" per_tool = _LONG_RUNNING_TOOL_TIMEOUTS.get(tool) if per_tool is not None: env_key, default_ms = per_tool return _timeout_from_env(env_key, default_ms) return _timeout_from_env("RSTACK_BRIDGE_TIMEOUT_MS", _CONTROL_PLANE_TIMEOUT_MS) async def _run_bridge(tool: str, params: dict, cwd: str, invocation_id: str, config_env: dict[str, str]) -> ToolResult: # G-09 (#552): prefer the locally-installed rstack-bridge bin (no npx # wrapper process, no cold registry fetch) over `npx --yes`; the npx # fallback keeps the zero-install path working — see resolve.py. argv_prefix, _needs_network = _resolve_bin_argv("rstack-bridge", cwd) if argv_prefix is None: return ToolResult.error(invocation_id, "RStack: no rstack-bridge binary and no `npx` on PATH. Install Node.js, then run `npm install rstack-agents` in this project.") env = {**os.environ, **config_env, "RSTACK_PROJECT_ROOT": cwd, "RSTACK_BRIDGE_CALLER": "tau"} try: proc = await asyncio.create_subprocess_exec( *argv_prefix, tool, json.dumps(params), cwd=cwd, env=env, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, **_own_process_group_kwargs(), ) except OSError as exc: # #487 (CodeRabbit review): _run_guard already catches this; a rare # spawn failure (permission error, resource limit) between the # shutil.which check above and exec should return a normal tool # error, not propagate an uncaught exception out of this tool call. return ToolResult.error(invocation_id, f"RStack: could not start the bridge process: {exc}") # #487 (audit finding, verbatim): "_run_bridge — the adapter's primary # tool-call path... A stalled `npx --yes rstack-bridge` (cold registry # fetch, network hiccup) blocks that call — and therefore the host — # indefinitely." Bounded + reaped via the shared helper. timeout_env = _LONG_RUNNING_TOOL_TIMEOUTS.get(tool, ("RSTACK_BRIDGE_TIMEOUT_MS", 0))[0] try: out, err = await _communicate_or_kill(proc, None, _bridge_timeout_s(tool)) except asyncio.TimeoutError: return ToolResult.error(invocation_id, f"RStack {tool} timed out after {_bridge_timeout_s(tool):.0f}s ({timeout_env}) — the bridge process was killed. This is a timeout, not a tool failure; a stalled npx/registry fetch or a hung bridge process is the likely cause.") stdout = out.decode("utf-8", "replace").strip() stderr = err.decode("utf-8", "replace").strip() if proc.returncode != 0: detail = stderr or stdout or f"exit {proc.returncode}" return ToolResult.error(invocation_id, f"RStack {tool} failed: {detail}") text = _extract_text(stdout) return ToolResult.ok(invocation_id, text) def _extract_text(stdout: str) -> str: """The bridge prints the tool's raw result. Pi tools return { content: [{type:'text', text}], details }. Pull the text out; fall back to raw stdout if the shape is unexpected.""" if not stdout: return "" try: data = json.loads(stdout) except json.JSONDecodeError: return stdout if isinstance(data, dict): content = data.get("content") if isinstance(content, list): parts = [str(c.get("text", "")) for c in content if isinstance(c, dict)] joined = "\n".join(p for p in parts if p) if joined: return joined return json.dumps(data, indent=2) return stdout class _BridgeTool(Tool): """One `sdlc_*` tool; `execute` shells out to the shared Node bridge.""" def __init__(self, name: str, description: str, schema, config_env: dict[str, str]): super().__init__( name=name, description=description, schema=schema, kind=ToolKind.Execute, execution_mode=ToolExecutionMode.Sequential, ) self._config_env = config_env async def execute( self, invocation: ToolInvocation, tool_execution_update_callback=None, signal=None, context: Optional[ToolContext] = None, ) -> ToolResult: cwd = str(getattr(context, "cwd", None) or os.getcwd()) params = {k: v for k, v in (invocation.params or {}).items() if v is not None} return await _run_bridge(self.name, params, cwd, invocation.id, self._config_env)