diff --git a/examples/agents/16k_credentials_google_adk.py b/examples/agents/16k_credentials_google_adk.py index ef0415cc..51112c1c 100644 --- a/examples/agents/16k_credentials_google_adk.py +++ b/examples/agents/16k_credentials_google_adk.py @@ -19,21 +19,22 @@ from conductor.ai.agents import AgentRuntime +def check_github_auth() -> str: + """Check if GitHub authentication is available.""" + token = os.environ.get("GITHUB_TOKEN", "") + if token: + return f"GitHub token is set (starts with {token[:4]}...)" + return "GitHub token is NOT set" + + def create_adk_agent(): """Create a Google ADK agent with a credential-aware tool.""" from google.adk import Agent from google.adk.tools import FunctionTool - def check_github_auth() -> str: - """Check if GitHub authentication is available.""" - token = os.environ.get("GITHUB_TOKEN", "") - if token: - return f"GitHub token is set (starts with {token[:4]}...)" - return "GitHub token is NOT set" - agent = Agent( name="github_checker", - model="gemini-2.5-flash", + model="gemini-3.6-flash", instruction="You check GitHub authentication status.", tools=[FunctionTool(check_github_auth)], ) diff --git a/examples/agents/79_agent_message_bus.py b/examples/agents/79_agent_message_bus.py index 554f1b3c..f2dde389 100644 --- a/examples/agents/79_agent_message_bus.py +++ b/examples/agents/79_agent_message_bus.py @@ -42,11 +42,14 @@ from conductor.ai.agents import Agent, AgentRuntime, tool, wait_for_message_tool from settings import settings -# Shared directory for IPC between main process and worker processes. -# Workers run as separate OS processes (different PIDs, same filesystem). -_ipc_dir = Path(tempfile.mkdtemp(prefix="message_bus_")) +_IPC_DIR_ENV = "MESSAGE_BUS_IPC_DIR" +if _IPC_DIR_ENV in os.environ: + _ipc_dir = Path(os.environ[_IPC_DIR_ENV]) +else: + _ipc_dir = Path(tempfile.mkdtemp(prefix="message_bus_")) + os.environ[_IPC_DIR_ENV] = str(_ipc_dir) _FORWARDED_DIR = _ipc_dir / "forwarded" # one file per forwarded topic -_FORWARDED_DIR.mkdir() +_FORWARDED_DIR.mkdir(exist_ok=True) TOPICS = [ "the impact of edge computing on cloud infrastructure", @@ -54,8 +57,21 @@ "how vector databases work", ] +_WRITER_EXECUTION_ID_ENV = "MESSAGE_BUS_WRITER_EXECUTION_ID" -def build_researcher(runtime: AgentRuntime, writer_execution_id: str) -> Agent: + +@tool +def forward_to_writer(topic: str, notes: str) -> str: + """Forward research notes to the Writer and signal the main process.""" + print(f" [researcher → writer] forwarding notes on {topic!r}") + writer_execution_id = os.environ[_WRITER_EXECUTION_ID_ENV] + with AgentRuntime() as rt: + rt.send_message(writer_execution_id, {"topic": topic, "notes": notes}) + (_FORWARDED_DIR / f"{time.time_ns()}.done").touch() + return "forwarded" + + +def build_researcher() -> Agent: """Build the Researcher agent with a forward tool wired to the Writer's queue.""" receive_topic = wait_for_message_tool( @@ -63,14 +79,6 @@ def build_researcher(runtime: AgentRuntime, writer_execution_id: str) -> Agent: description="Wait for the next research topic.", ) - @tool - def forward_to_writer(topic: str, notes: str) -> str: - """Forward research notes to the Writer and signal the main process.""" - print(f" [researcher → writer] forwarding notes on {topic!r}") - runtime.send_message(writer_execution_id, {"topic": topic, "notes": notes}) - (_FORWARDED_DIR / f"{time.time_ns()}.done").touch() - return "forwarded" - return Agent( name="researcher", model=settings.llm_model, @@ -88,6 +96,14 @@ def forward_to_writer(topic: str, notes: str) -> str: ) +@tool +def publish(topic: str, paragraph: str) -> str: + """Publish the finished paragraph.""" + print(f"\n [writer] ── {topic} ──") + print(f" {paragraph}\n") + return "published" + + def build_writer() -> Agent: """Build the Writer agent that polishes research notes into paragraphs.""" @@ -99,13 +115,6 @@ def build_writer() -> Agent: ), ) - @tool - def publish(topic: str, paragraph: str) -> str: - """Publish the finished paragraph.""" - print(f"\n [writer] ── {topic} ──") - print(f" {paragraph}\n") - return "published" - return Agent( name="writer", model=settings.llm_model, @@ -122,34 +131,41 @@ def publish(topic: str, paragraph: str) -> str: ) -try: - with AgentRuntime() as runtime: - # Start the Writer first so its execution_id is available to the Researcher - writer_handle = runtime.start(build_writer(), "Begin. Wait for research notes.") - writer_id = writer_handle.execution_id - print(f"Writer started: {writer_id}") - - researcher = build_researcher(runtime, writer_id) - researcher_handle = runtime.start(researcher, "Begin. Wait for your first topic.") - researcher_id = researcher_handle.execution_id - print(f"Researcher started: {researcher_id}\n") - - time.sleep(4) - print("Sending topics to Researcher...\n") - for topic in TOPICS: - print(f" → {topic!r}") - runtime.send_message(researcher_id, {"topic": topic}) - - # Wait until all topics have been forwarded to the Writer - while len(list(_FORWARDED_DIR.iterdir())) < len(TOPICS): - time.sleep(0.1) - - # Deterministic stop — no stop-handling instructions needed. - researcher_handle.stop() - writer_handle.stop() - researcher_handle.join(timeout=30) - writer_handle.join(timeout=30) - - print("Done.") -finally: - shutil.rmtree(_ipc_dir, ignore_errors=True) +def main() -> None: + try: + with AgentRuntime() as runtime: + # Start the Writer first so its execution_id is available to the Researcher + writer_handle = runtime.start(build_writer(), "Begin. Wait for research notes.") + writer_id = writer_handle.execution_id + print(f"Writer started: {writer_id}") + + os.environ[_WRITER_EXECUTION_ID_ENV] = writer_id + + researcher = build_researcher() + researcher_handle = runtime.start(researcher, "Begin. Wait for your first topic.") + researcher_id = researcher_handle.execution_id + print(f"Researcher started: {researcher_id}\n") + + time.sleep(4) + print("Sending topics to Researcher...\n") + for topic in TOPICS: + print(f" → {topic!r}") + runtime.send_message(researcher_id, {"topic": topic}) + + # Wait until all topics have been forwarded to the Writer + while len(list(_FORWARDED_DIR.iterdir())) < len(TOPICS): + time.sleep(0.1) + + # Deterministic stop — no stop-handling instructions needed. + researcher_handle.stop() + writer_handle.stop() + researcher_handle.join(timeout=30) + writer_handle.join(timeout=30) + + print("Done.") + finally: + shutil.rmtree(_ipc_dir, ignore_errors=True) + + +if __name__ == "__main__": + main() diff --git a/examples/agents/80_live_dashboard.py b/examples/agents/80_live_dashboard.py index 8840383f..cdabaefb 100644 --- a/examples/agents/80_live_dashboard.py +++ b/examples/agents/80_live_dashboard.py @@ -55,12 +55,16 @@ from conductor.ai.agents import Agent, AgentRuntime, tool, wait_for_message_tool -# Filesystem IPC between main process and worker processes (separate OS PIDs). -_ipc_dir = Path(tempfile.mkdtemp(prefix="live_dashboard_")) +_IPC_DIR_ENV = "LIVE_DASHBOARD_IPC_DIR" +if _IPC_DIR_ENV in os.environ: + _ipc_dir = Path(os.environ[_IPC_DIR_ENV]) +else: + _ipc_dir = Path(tempfile.mkdtemp(prefix="live_dashboard_")) + os.environ[_IPC_DIR_ENV] = str(_ipc_dir) _BATCH_DIR = _ipc_dir / "batches" # one file per batch dispatched by Feeder _DISPLAY_DIR = _ipc_dir / "displays" # one file per display_dashboard call by Monitor -_BATCH_DIR.mkdir() -_DISPLAY_DIR.mkdir() +_BATCH_DIR.mkdir(exist_ok=True) +_DISPLAY_DIR.mkdir(exist_ok=True) _MONITOR_ID_FILE = _ipc_dir / "monitor_id.txt" # written by main, read by Feeder tool @@ -68,6 +72,18 @@ # Monitor agent # --------------------------------------------------------------------------- +@tool +def display_dashboard(summary: str) -> str: + """Publish an aggregated dashboard line for this batch. + + Writes the summary to a file in _DISPLAY_DIR so the main process can + read and print it. The file name encodes arrival order via time_ns. + """ + ts = time.time_ns() + (_DISPLAY_DIR / f"{ts}.txt").write_text(summary) + return "displayed" + + def build_monitor() -> Agent: """Monitor: pulls up to 10 metrics per call and prints aggregated stats.""" @@ -80,17 +96,6 @@ def build_monitor() -> Agent: batch_size=10, ) - @tool - def display_dashboard(summary: str) -> str: - """Publish an aggregated dashboard line for this batch. - - Writes the summary to a file in _DISPLAY_DIR so the main process can - read and print it. The file name encodes arrival order via time_ns. - """ - ts = time.time_ns() - (_DISPLAY_DIR / f"{ts}.txt").write_text(summary) - return "displayed" - return Agent( name="monitor_agent", model=settings.llm_model, @@ -114,41 +119,43 @@ def display_dashboard(summary: str) -> str: # Feeder agent # --------------------------------------------------------------------------- -def build_feeder(runtime: AgentRuntime) -> Agent: - """Feeder: generates metric samples and pushes them into the Monitor's queue.""" - - receive_signal = wait_for_message_tool( - name="receive_signal", - description="Wait for a control signal from the orchestrator ({batches: N}).", - ) - - @tool - def push_metrics_batch(batch_number: int) -> str: - """Generate and push one batch of metric samples to the Monitor agent. - - Reads the Monitor's execution ID from a shared file and sends 5 metric - samples directly into its WMQ. Writes a sentinel file so the main - process knows the batch was dispatched. - """ - monitor_id = _MONITOR_ID_FILE.read_text().strip() - metrics = [ - "cpu_pct", - "mem_mb", - "req_rate", - "latency_ms", - "error_rate", - ] - samples = [] +@tool +def push_metrics_batch(batch_number: int) -> str: + """Generate and push one batch of metric samples to the Monitor agent. + + Reads the Monitor's execution ID from a shared file and sends 5 metric + samples directly into its WMQ. Writes a sentinel file so the main + process knows the batch was dispatched. + """ + monitor_id = _MONITOR_ID_FILE.read_text().strip() + metrics = [ + "cpu_pct", + "mem_mb", + "req_rate", + "latency_ms", + "error_rate", + ] + samples = [] + with AgentRuntime() as rt: for _ in range(5): metric = random.choice(metrics) host = random.choice(["web-01", "web-02", "db-01"]) value = round(random.uniform(0, 100), 2) sample = {"metric": metric, "host": host, "value": value} samples.append(sample) - runtime.send_message(monitor_id, sample) + rt.send_message(monitor_id, sample) + + (_BATCH_DIR / f"batch_{batch_number}_{time.time_ns()}.done").touch() + return f"Pushed {len(samples)} samples in batch {batch_number}: {json.dumps(samples)}" + + +def build_feeder() -> Agent: + """Feeder: generates metric samples and pushes them into the Monitor's queue.""" - (_BATCH_DIR / f"batch_{batch_number}_{time.time_ns()}.done").touch() - return f"Pushed {len(samples)} samples in batch {batch_number}: {json.dumps(samples)}" + receive_signal = wait_for_message_tool( + name="receive_signal", + description="Wait for a control signal from the orchestrator ({batches: N}).", + ) return Agent( name="feeder_agent", @@ -176,54 +183,59 @@ def push_metrics_batch(batch_number: int) -> str: # How many display_dashboard calls to expect before sending stop: EXPECTED_DISPLAYS = math.ceil(TOTAL_BATCHES * SAMPLES_PER_BATCH / MONITOR_BATCH_SIZE) -try: - with AgentRuntime() as runtime: - # Start Monitor first so its execution_id exists before Feeder needs it. - monitor_handle = runtime.start(build_monitor(), "Begin. Wait for metric batches.") - monitor_id = monitor_handle.execution_id - _MONITOR_ID_FILE.write_text(monitor_id) - print(f"Monitor started: {monitor_id}") - - feeder_handle = runtime.start(build_feeder(runtime), "Begin. Wait for orchestrator signals.") - feeder_id = feeder_handle.execution_id - print(f"Feeder started: {feeder_id}\n") - - # Give agents time to reach their first wait_for_message call. - time.sleep(4) - - print(f"Sending {TOTAL_BATCHES} batch signals to Feeder (5 metrics each = " - f"{TOTAL_BATCHES * 5} total samples, Monitor reads ≤10 per call)...\n") - - # Send batch signals two at a time to let the Feeder bundle them. - runtime.send_message(feeder_id, {"batches": TOTAL_BATCHES // 2}) - runtime.send_message(feeder_id, {"batches": TOTAL_BATCHES - TOTAL_BATCHES // 2}) - - # Wait until all batches have been dispatched via push_metrics_batch. - print("Waiting for all batches to be dispatched...") - while len(list(_BATCH_DIR.iterdir())) < TOTAL_BATCHES: - time.sleep(0.1) - print(f" All {TOTAL_BATCHES} batches dispatched ({TOTAL_BATCHES * SAMPLES_PER_BATCH} samples in Monitor's queue).\n") - - # Tail _DISPLAY_DIR: print summaries as they arrive, wait until all done. - # Without this barrier, AgentRuntime.__exit__ kills the display_dashboard - # worker while Monitor's LLM is still pulling batches from the queue. - print(f"Live dashboard (Monitor processes ≤{MONITOR_BATCH_SIZE} samples per batch):\n") - seen: set[str] = set() - batch_index = 0 - while len(seen) < EXPECTED_DISPLAYS: - for p in sorted(_DISPLAY_DIR.iterdir()): - if p.name not in seen and p.suffix == ".txt": - batch_index += 1 - print(f" [dashboard batch {batch_index}] {p.read_text()}") - seen.add(p.name) - time.sleep(0.05) - - print(f"\nAll {EXPECTED_DISPLAYS} batch reports received. Stopping...\n") - feeder_handle.stop() - monitor_handle.stop() - feeder_handle.join(timeout=30) - monitor_handle.join(timeout=30) - - print("Done.") -finally: - shutil.rmtree(_ipc_dir, ignore_errors=True) +def main() -> None: + try: + with AgentRuntime() as runtime: + # Start Monitor first so its execution_id exists before Feeder needs it. + monitor_handle = runtime.start(build_monitor(), "Begin. Wait for metric batches.") + monitor_id = monitor_handle.execution_id + _MONITOR_ID_FILE.write_text(monitor_id) + print(f"Monitor started: {monitor_id}") + + feeder_handle = runtime.start(build_feeder(), "Begin. Wait for orchestrator signals.") + feeder_id = feeder_handle.execution_id + print(f"Feeder started: {feeder_id}\n") + + # Give agents time to reach their first wait_for_message call. + time.sleep(4) + + print(f"Sending {TOTAL_BATCHES} batch signals to Feeder (5 metrics each = " + f"{TOTAL_BATCHES * 5} total samples, Monitor reads ≤10 per call)...\n") + + # Send batch signals two at a time to let the Feeder bundle them. + runtime.send_message(feeder_id, {"batches": TOTAL_BATCHES // 2}) + runtime.send_message(feeder_id, {"batches": TOTAL_BATCHES - TOTAL_BATCHES // 2}) + + # Wait until all batches have been dispatched via push_metrics_batch. + print("Waiting for all batches to be dispatched...") + while len(list(_BATCH_DIR.iterdir())) < TOTAL_BATCHES: + time.sleep(0.1) + print(f" All {TOTAL_BATCHES} batches dispatched ({TOTAL_BATCHES * SAMPLES_PER_BATCH} samples in Monitor's queue).\n") + + # Tail _DISPLAY_DIR: print summaries as they arrive, wait until all done. + # Without this barrier, AgentRuntime.__exit__ kills the display_dashboard + # worker while Monitor's LLM is still pulling batches from the queue. + print(f"Live dashboard (Monitor processes ≤{MONITOR_BATCH_SIZE} samples per batch):\n") + seen: set[str] = set() + batch_index = 0 + while len(seen) < EXPECTED_DISPLAYS: + for p in sorted(_DISPLAY_DIR.iterdir()): + if p.name not in seen and p.suffix == ".txt": + batch_index += 1 + print(f" [dashboard batch {batch_index}] {p.read_text()}") + seen.add(p.name) + time.sleep(0.05) + + print(f"\nAll {EXPECTED_DISPLAYS} batch reports received. Stopping...\n") + feeder_handle.stop() + monitor_handle.stop() + feeder_handle.join(timeout=30) + monitor_handle.join(timeout=30) + + print("Done.") + finally: + shutil.rmtree(_ipc_dir, ignore_errors=True) + + +if __name__ == "__main__": + main() diff --git a/examples/agents/81_chat_repl.py b/examples/agents/81_chat_repl.py index 4dcf4a4f..8c54170a 100644 --- a/examples/agents/81_chat_repl.py +++ b/examples/agents/81_chat_repl.py @@ -97,7 +97,12 @@ # Filesystem IPC setup # --------------------------------------------------------------------------- -_ipc_dir = Path(tempfile.mkdtemp(prefix="chat_repl_")) +_IPC_DIR_ENV = "CHAT_REPL_IPC_DIR" +if _IPC_DIR_ENV in os.environ: + _ipc_dir = Path(os.environ[_IPC_DIR_ENV]) +else: + _ipc_dir = Path(tempfile.mkdtemp(prefix="chat_repl_")) + os.environ[_IPC_DIR_ENV] = str(_ipc_dir) _REPLY_FILE = _ipc_dir / "reply.txt" # agent writes reply here _REPLY_READY = _ipc_dir / "reply.ready" # sentinel: reply is ready to read _REGISTRY_FILE = _ipc_dir / "registry.json" # active ephemeral tasks @@ -119,6 +124,40 @@ def _read_registry() -> dict: # Agent definition # --------------------------------------------------------------------------- +@tool +def reply_to_user(message: str) -> str: + """Send a reply back to the user in the REPL. + + Writes the reply to a shared file and touches a sentinel so the main + thread knows a new reply is ready to display. + """ + _REPLY_FILE.write_text(message) + _REPLY_READY.touch() + return "reply sent" + + +@tool +def run_task(task_name: str, task_input: str) -> str: + """Run a registered ephemeral task by name. + + Reads the active task registry at call time — newly registered tasks + are available immediately. Returns the task output or an error if the + task name is not registered. + """ + registry = _read_registry() + if task_name not in registry: + available = ", ".join(registry) or "(none)" + return f"Error: task '{task_name}' not found. Available: {available}" + impl_fn = _TASK_IMPLEMENTATIONS.get(task_name) + if impl_fn is None: + return f"Error: task '{task_name}' has no implementation." + _, fn = impl_fn + try: + return fn(task_input) + except Exception as exc: + return f"Error running '{task_name}': {exc}" + + def build_agent() -> Agent: receive_message = wait_for_message_tool( name="wait_for_message", @@ -129,38 +168,6 @@ def build_agent() -> Agent: ), ) - @tool - def reply_to_user(message: str) -> str: - """Send a reply back to the user in the REPL. - - Writes the reply to a shared file and touches a sentinel so the main - thread knows a new reply is ready to display. - """ - _REPLY_FILE.write_text(message) - _REPLY_READY.touch() - return "reply sent" - - @tool - def run_task(task_name: str, task_input: str) -> str: - """Run a registered ephemeral task by name. - - Reads the active task registry at call time — newly registered tasks - are available immediately. Returns the task output or an error if the - task name is not registered. - """ - registry = _read_registry() - if task_name not in registry: - available = ", ".join(registry) or "(none)" - return f"Error: task '{task_name}' not found. Available: {available}" - impl_fn = _TASK_IMPLEMENTATIONS.get(task_name) - if impl_fn is None: - return f"Error: task '{task_name}' has no implementation." - _, fn = impl_fn - try: - return fn(task_input) - except Exception as exc: - return f"Error running '{task_name}': {exc}" - return Agent( name="chat_repl_agent", model=settings.llm_model, @@ -224,99 +231,104 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() -try: - args = parse_args() - active_tasks: dict[str, str] = {} - _write_registry(active_tasks) - agent = build_agent() - - with AgentRuntime() as runtime: - if args.resume: - if not args.session_file.exists(): - print(f"No session file found at {args.session_file}") - print("Start a new session first (without --resume).") - raise SystemExit(1) - - saved_eid = args.session_file.read_text().strip() - print(f"Resuming session: {saved_eid}") - - # resume() fetches the workflow from the server, extracts the - # domain from taskToDomain, and re-registers workers under it. - handle = runtime.resume(saved_eid, agent) - execution_id = handle.execution_id - print(f"Workers re-registered under domain: {handle.run_id}") - else: - handle = runtime.start(agent, "Begin. Wait for the user's first message.") - execution_id = handle.execution_id - args.session_file.write_text(execution_id) - print(f"Agent started: {execution_id}") - print(f"Domain (run_id): {handle.run_id}") - print(f"Session saved to {args.session_file}") - - print("\n" + "=" * 60) - print("Chat REPL — type 'help' for commands, 'quit' to exit") - print("=" * 60 + "\n") - - while True: - try: - user_input = input("You: ").strip() - except (EOFError, KeyboardInterrupt): - print("\n\nDisconnected (Ctrl+C). Resume later with --resume.") - break - - if not user_input: - continue - - if user_input.lower() in ("quit", "exit"): - handle.stop() - print("Agent stopped.\n") - # Clean up session file — agent is stopped - if args.session_file.exists(): - args.session_file.unlink() - break - - if user_input.lower() == "/disconnect": - print("Disconnected. Resume later with: python 81_chat_repl.py --resume") - break - - if user_input.lower() == "help": - print(HELP_TEXT) - continue - - if user_input.lower() == "/tools": - if active_tasks: - print("Active ephemeral tasks:") - for name, desc in active_tasks.items(): - print(f" {name:12s} {desc}") - else: - print("No ephemeral tasks activated yet. Use /tool .") - print() - continue - - if user_input.lower().startswith("/tool "): - task_name = user_input[6:].strip() - if task_name not in _TASK_IMPLEMENTATIONS: - print(f"Unknown task '{task_name}'. " - f"Available: {', '.join(_TASK_IMPLEMENTATIONS)}\n") +def main() -> None: + try: + args = parse_args() + active_tasks: dict[str, str] = {} + _write_registry(active_tasks) + agent = build_agent() + + with AgentRuntime() as runtime: + if args.resume: + if not args.session_file.exists(): + print(f"No session file found at {args.session_file}") + print("Start a new session first (without --resume).") + raise SystemExit(1) + + saved_eid = args.session_file.read_text().strip() + print(f"Resuming session: {saved_eid}") + + # resume() fetches the workflow from the server, extracts the + # domain from taskToDomain, and re-registers workers under it. + handle = runtime.resume(saved_eid, agent) + execution_id = handle.execution_id + print(f"Workers re-registered under domain: {handle.run_id}") + else: + handle = runtime.start(agent, "Begin. Wait for the user's first message.") + execution_id = handle.execution_id + args.session_file.write_text(execution_id) + print(f"Agent started: {execution_id}") + print(f"Domain (run_id): {handle.run_id}") + print(f"Session saved to {args.session_file}") + + print("\n" + "=" * 60) + print("Chat REPL — type 'help' for commands, 'quit' to exit") + print("=" * 60 + "\n") + + while True: + try: + user_input = input("You: ").strip() + except (EOFError, KeyboardInterrupt): + print("\n\nDisconnected (Ctrl+C). Resume later with --resume.") + break + + if not user_input: + continue + + if user_input.lower() in ("quit", "exit"): + handle.stop() + print("Agent stopped.\n") + # Clean up session file — agent is stopped + if args.session_file.exists(): + args.session_file.unlink() + break + + if user_input.lower() == "/disconnect": + print("Disconnected. Resume later with: python 81_chat_repl.py --resume") + break + + if user_input.lower() == "help": + print(HELP_TEXT) continue - desc, _ = _TASK_IMPLEMENTATIONS[task_name] - active_tasks[task_name] = desc - _write_registry(active_tasks) - print(f" → Registered ephemeral task '{task_name}'.\n") - # Notify the agent so it can acknowledge and use it in the next turn. - runtime.send_message(execution_id, { - "tool_registered": task_name, - "tool_description": desc, - }) + + if user_input.lower() == "/tools": + if active_tasks: + print("Active ephemeral tasks:") + for name, desc in active_tasks.items(): + print(f" {name:12s} {desc}") + else: + print("No ephemeral tasks activated yet. Use /tool .") + print() + continue + + if user_input.lower().startswith("/tool "): + task_name = user_input[6:].strip() + if task_name not in _TASK_IMPLEMENTATIONS: + print(f"Unknown task '{task_name}'. " + f"Available: {', '.join(_TASK_IMPLEMENTATIONS)}\n") + continue + desc, _ = _TASK_IMPLEMENTATIONS[task_name] + active_tasks[task_name] = desc + _write_registry(active_tasks) + print(f" → Registered ephemeral task '{task_name}'.\n") + # Notify the agent so it can acknowledge and use it in the next turn. + runtime.send_message(execution_id, { + "tool_registered": task_name, + "tool_description": desc, + }) + reply = _wait_for_reply() + print(f"Agent: {reply}\n") + continue + + # Normal user message. + runtime.send_message(execution_id, {"text": user_input}) reply = _wait_for_reply() print(f"Agent: {reply}\n") - continue - # Normal user message. - runtime.send_message(execution_id, {"text": user_input}) - reply = _wait_for_reply() - print(f"Agent: {reply}\n") + print("Session ended.") + finally: + shutil.rmtree(_ipc_dir, ignore_errors=True) + - print("Session ended.") -finally: - shutil.rmtree(_ipc_dir, ignore_errors=True) +if __name__ == "__main__": + main() diff --git a/examples/agents/82_coding_agent.py b/examples/agents/82_coding_agent.py index e29a7fbd..9ce43de3 100644 --- a/examples/agents/82_coding_agent.py +++ b/examples/agents/82_coding_agent.py @@ -43,6 +43,17 @@ _MAX_SHELL_OUTPUT = 8_000 # truncate shell output shown to the LLM _MAX_SHELL_DISPLAY = 2_000 # truncate shell output shown in the terminal +_WORKING_DIR_ENV = "CODING_AGENT_WORKING_DIR" +_SHELL_TIMEOUT_ENV = "CODING_AGENT_SHELL_TIMEOUT" + + +def _working_dir() -> str: + return os.environ[_WORKING_DIR_ENV] + + +def _shell_timeout() -> int: + return int(os.environ.get(_SHELL_TIMEOUT_ENV, str(_DEFAULT_SHELL_TIMEOUT))) + # --------------------------------------------------------------------------- # Terminal display @@ -266,149 +277,165 @@ def _stream_events() -> None: # Agent builder # --------------------------------------------------------------------------- +@tool +def read_file(path: str) -> str: + """Read a file and return its text contents. Paths may be absolute or relative to the working directory.""" + working_dir = _working_dir() + target = Path(path) if os.path.isabs(path) else Path(working_dir) / path + if not target.exists(): + return f"Error: {path!r} does not exist." + if target.is_dir(): + return f"Error: {path!r} is a directory. Use list_dir to browse it." + size = target.stat().st_size + if size > _MAX_FILE_BYTES: + return ( + f"Error: {path!r} is {size:,} bytes (limit {_MAX_FILE_BYTES:,}). " + "Use search_in_files to find specific content instead." + ) + try: + return target.read_text(encoding="utf-8", errors="replace") + except Exception as exc: + return f"Error reading {path!r}: {exc}" + + +@tool +def write_file(path: str, content: str) -> str: + """Write content to a file, creating parent directories as needed. Overwrites existing files.""" + working_dir = _working_dir() + target = Path(path) if os.path.isabs(path) else Path(working_dir) / path + try: + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + return f"Wrote {len(content):,} bytes to {str(target)!r}." + except Exception as exc: + return f"Error writing {path!r}: {exc}" + + +@tool +def list_dir(path: str = ".") -> str: + """List directory contents with file sizes. Paths may be absolute or relative to the working directory.""" + working_dir = _working_dir() + target = Path(path) if os.path.isabs(path) else Path(working_dir) / path + if not target.exists(): + return f"Error: {path!r} does not exist." + if not target.is_dir(): + return f"Error: {path!r} is not a directory." + try: + entries = sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name)) + lines = [] + for entry in entries: + if entry.is_dir(): + lines.append(f" {entry.name}/") + else: + lines.append(f" {entry.name} ({entry.stat().st_size:,} bytes)") + header = str(target) + "/" + return header + "\n" + "\n".join(lines) if lines else header + " (empty)" + except Exception as exc: + return f"Error listing {path!r}: {exc}" + + +@tool +def find_files(pattern: str, path: str = ".") -> str: + """Find files matching a glob pattern (e.g. '**/*.py'). Path relative to working directory.""" + working_dir = _working_dir() + base = Path(path) if os.path.isabs(path) else Path(working_dir) / path + if not base.exists(): + return f"Error: {path!r} does not exist." + if not base.is_dir(): + return f"Error: {path!r} is not a directory." + try: + matches = sorted(m for m in base.glob(pattern) if m.is_file()) + if not matches: + return f"No files matching {pattern!r} under {str(base)!r}." + lines = [] + for m in matches[:200]: + try: + rel = m.relative_to(working_dir) + except ValueError: + rel = m + lines.append(str(rel)) + suffix = f"\n... ({len(matches) - 200} more)" if len(matches) > 200 else "" + return "\n".join(lines) + suffix + except Exception as exc: + return f"Error finding files: {exc}" + + +@tool +def search_in_files(regex: str, path: str = ".", file_glob: str = "**/*") -> str: + """Search for a regex pattern in file contents. Returns file:line: matching_line entries.""" + import re as _re + working_dir = _working_dir() + base = Path(path) if os.path.isabs(path) else Path(working_dir) / path + try: + compiled = _re.compile(regex) + except _re.error as exc: + return f"Invalid regex {regex!r}: {exc}" + results = [] + for filepath in sorted(base.glob(file_glob)): + if not filepath.is_file() or filepath.stat().st_size > _MAX_FILE_BYTES: + continue + try: + for lineno, line in enumerate( + filepath.read_text(encoding="utf-8", errors="replace").splitlines(), 1 + ): + if compiled.search(line): + try: + label = str(filepath.relative_to(working_dir)) + except ValueError: + label = str(filepath) + results.append(f"{label}:{lineno}: {line.rstrip()}") + if len(results) >= 100: + break + except Exception: + continue + if len(results) >= 100: + break + if not results: + return f"No matches for {regex!r} in {str(base)!r} ({file_glob})." + suffix = "\n... (truncated at 100 matches)" if len(results) >= 100 else "" + return "\n".join(results) + suffix + + +@tool +def run_shell(command: str) -> str: + """Run a shell command in the working directory. Returns stdout + stderr with exit code.""" + working_dir = _working_dir() + shell_timeout = _shell_timeout() + try: + proc = subprocess.run( + command, + shell=True, + cwd=working_dir, + capture_output=True, + text=True, + timeout=shell_timeout, + ) + combined = (proc.stdout + proc.stderr).strip() + if len(combined) > _MAX_SHELL_OUTPUT: + combined = combined[:_MAX_SHELL_OUTPUT] + f"\n... (truncated, {len(combined):,} chars total)" + return f"[exit {proc.returncode}]\n{combined}" if combined else f"[exit {proc.returncode}] (no output)" + except subprocess.TimeoutExpired: + return f"Error: command timed out after {shell_timeout}s." + except Exception as exc: + return f"Error: {exc}" + + +@tool +def reply_to_user(message: str) -> str: + """Send your response to the user. Call this when the task is complete.""" + return "ok" + + def build_agent(working_dir: str, shell_timeout: int = _DEFAULT_SHELL_TIMEOUT) -> Agent: - """Build the coding agent. All tools close over working_dir and shell_timeout.""" + """Build the coding agent. Tools read working_dir/shell_timeout from env vars set here.""" + os.environ[_WORKING_DIR_ENV] = working_dir + os.environ[_SHELL_TIMEOUT_ENV] = str(shell_timeout) receive_message = wait_for_message_tool( name="wait_for_message", description="Wait for the next user message. Payload has a 'text' field.", ) - @tool - def read_file(path: str) -> str: - """Read a file and return its text contents. Paths may be absolute or relative to the working directory.""" - target = Path(path) if os.path.isabs(path) else Path(working_dir) / path - if not target.exists(): - return f"Error: {path!r} does not exist." - if target.is_dir(): - return f"Error: {path!r} is a directory. Use list_dir to browse it." - size = target.stat().st_size - if size > _MAX_FILE_BYTES: - return ( - f"Error: {path!r} is {size:,} bytes (limit {_MAX_FILE_BYTES:,}). " - "Use search_in_files to find specific content instead." - ) - try: - return target.read_text(encoding="utf-8", errors="replace") - except Exception as exc: - return f"Error reading {path!r}: {exc}" - - @tool - def write_file(path: str, content: str) -> str: - """Write content to a file, creating parent directories as needed. Overwrites existing files.""" - target = Path(path) if os.path.isabs(path) else Path(working_dir) / path - try: - target.parent.mkdir(parents=True, exist_ok=True) - target.write_text(content, encoding="utf-8") - return f"Wrote {len(content):,} bytes to {str(target)!r}." - except Exception as exc: - return f"Error writing {path!r}: {exc}" - - @tool - def list_dir(path: str = ".") -> str: - """List directory contents with file sizes. Paths may be absolute or relative to the working directory.""" - target = Path(path) if os.path.isabs(path) else Path(working_dir) / path - if not target.exists(): - return f"Error: {path!r} does not exist." - if not target.is_dir(): - return f"Error: {path!r} is not a directory." - try: - entries = sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name)) - lines = [] - for entry in entries: - if entry.is_dir(): - lines.append(f" {entry.name}/") - else: - lines.append(f" {entry.name} ({entry.stat().st_size:,} bytes)") - header = str(target) + "/" - return header + "\n" + "\n".join(lines) if lines else header + " (empty)" - except Exception as exc: - return f"Error listing {path!r}: {exc}" - - @tool - def find_files(pattern: str, path: str = ".") -> str: - """Find files matching a glob pattern (e.g. '**/*.py'). Path relative to working directory.""" - base = Path(path) if os.path.isabs(path) else Path(working_dir) / path - if not base.exists(): - return f"Error: {path!r} does not exist." - if not base.is_dir(): - return f"Error: {path!r} is not a directory." - try: - matches = sorted(m for m in base.glob(pattern) if m.is_file()) - if not matches: - return f"No files matching {pattern!r} under {str(base)!r}." - lines = [] - for m in matches[:200]: - try: - rel = m.relative_to(working_dir) - except ValueError: - rel = m - lines.append(str(rel)) - suffix = f"\n... ({len(matches) - 200} more)" if len(matches) > 200 else "" - return "\n".join(lines) + suffix - except Exception as exc: - return f"Error finding files: {exc}" - - @tool - def search_in_files(regex: str, path: str = ".", file_glob: str = "**/*") -> str: - """Search for a regex pattern in file contents. Returns file:line: matching_line entries.""" - import re as _re - base = Path(path) if os.path.isabs(path) else Path(working_dir) / path - try: - compiled = _re.compile(regex) - except _re.error as exc: - return f"Invalid regex {regex!r}: {exc}" - results = [] - for filepath in sorted(base.glob(file_glob)): - if not filepath.is_file() or filepath.stat().st_size > _MAX_FILE_BYTES: - continue - try: - for lineno, line in enumerate( - filepath.read_text(encoding="utf-8", errors="replace").splitlines(), 1 - ): - if compiled.search(line): - try: - label = str(filepath.relative_to(working_dir)) - except ValueError: - label = str(filepath) - results.append(f"{label}:{lineno}: {line.rstrip()}") - if len(results) >= 100: - break - except Exception: - continue - if len(results) >= 100: - break - if not results: - return f"No matches for {regex!r} in {str(base)!r} ({file_glob})." - suffix = "\n... (truncated at 100 matches)" if len(results) >= 100 else "" - return "\n".join(results) + suffix - - @tool - def run_shell(command: str) -> str: - """Run a shell command in the working directory. Returns stdout + stderr with exit code.""" - try: - proc = subprocess.run( - command, - shell=True, - cwd=working_dir, - capture_output=True, - text=True, - timeout=shell_timeout, - ) - combined = (proc.stdout + proc.stderr).strip() - if len(combined) > _MAX_SHELL_OUTPUT: - combined = combined[:_MAX_SHELL_OUTPUT] + f"\n... (truncated, {len(combined):,} chars total)" - return f"[exit {proc.returncode}]\n{combined}" if combined else f"[exit {proc.returncode}] (no output)" - except subprocess.TimeoutExpired: - return f"Error: command timed out after {shell_timeout}s." - except Exception as exc: - return f"Error: {exc}" - - @tool - def reply_to_user(message: str) -> str: - """Send your response to the user. Call this when the task is complete.""" - return "ok" - return Agent( name="coding_agent", model=settings.llm_model, diff --git a/examples/agents/82b_coding_agent_tui.py b/examples/agents/82b_coding_agent_tui.py index 70404193..1d9ee3f5 100644 --- a/examples/agents/82b_coding_agent_tui.py +++ b/examples/agents/82b_coding_agent_tui.py @@ -23,12 +23,15 @@ import argparse import enum +import json import os import queue +import shutil +import signal import subprocess +import tempfile import threading import time -from dataclasses import dataclass, field from pathlib import Path os.environ.setdefault("CONDUCTOR_LOG_LEVEL", "WARNING") @@ -51,11 +54,22 @@ _MAX_FILE_BYTES = 200_000 _MAX_SHELL_OUTPUT = 8_000 _MAX_SHELL_DISPLAY = 2_000 -_MAX_BG_BUFFER = 8_000 _SEPARATOR = "─" * 62 _THIN_SEP = "┄" * 62 +_WORKING_DIR_ENV = "CODING_AGENT_TUI_WORKING_DIR" +_SHELL_TIMEOUT_ENV = "CODING_AGENT_TUI_SHELL_TIMEOUT" +_BG_STATE_DIR_ENV = "CODING_AGENT_TUI_BG_DIR" + + +def _working_dir() -> str: + return os.environ[_WORKING_DIR_ENV] + + +def _shell_timeout() -> int: + return int(os.environ.get(_SHELL_TIMEOUT_ENV, str(_DEFAULT_SHELL_TIMEOUT))) + _HELP_TEXT = """\ Commands: @@ -90,123 +104,174 @@ class AgentState(enum.Enum): # Background process registry # --------------------------------------------------------------------------- -@dataclass -class BgProcess: - id: int - command: str - proc: subprocess.Popen - buffer: list = field(default_factory=list) - lock: threading.Lock = field(default_factory=threading.Lock) - started_at: float = field(default_factory=time.time) - _read_pos: int = field(default=0, repr=False) +def _bg_state_dir() -> Path: + d = Path(os.environ[_BG_STATE_DIR_ENV]) + d.mkdir(parents=True, exist_ok=True) + return d -def _start_reader_thread(bg: BgProcess) -> None: - """Daemon thread that reads stdout/stderr into the buffer.""" - def _read(): - try: - for line in bg.proc.stdout: - with bg.lock: - bg.buffer.append(line) - total = sum(len(ln) for ln in bg.buffer) - while total > _MAX_BG_BUFFER and len(bg.buffer) > 1: - total -= len(bg.buffer.pop(0)) - bg._read_pos = max(0, bg._read_pos - 1) - except Exception: - pass - threading.Thread(target=_read, daemon=True).start() +def _bg_registry_file() -> Path: + return _bg_state_dir() / "registry.json" + + +def _bg_log_file(bg_id: int) -> Path: + return _bg_state_dir() / f"{bg_id}.log" + + +def _bg_offset_file(bg_id: int) -> Path: + return _bg_state_dir() / f"{bg_id}.offset" + + +def _bg_exitcode_file(bg_id: int) -> Path: + return _bg_state_dir() / f"{bg_id}.exitcode" + + +def _read_bg_registry() -> dict: + f = _bg_registry_file() + if not f.exists(): + return {} + try: + return json.loads(f.read_text()) + except (json.JSONDecodeError, OSError): + return {} + +def _write_bg_registry(registry: dict) -> None: + _bg_registry_file().write_text(json.dumps(registry)) -def _make_bg_tools(working_dir: str): - """Create background process tools that close over a shared registry.""" - _bg_processes: dict[int, BgProcess] = {} - _next_id = [0] - @tool - def run_background(command: str) -> str: - """Start a long-running process in the background. Returns immediately with a process ID. - Use for servers, file watchers, builds — anything that won't exit quickly.""" - _next_id[0] += 1 - bg_id = _next_id[0] +def _bg_status(bg_id: int, pid: int) -> str: + """Running/exited status for a background process, probed from any process.""" + exitcode_file = _bg_exitcode_file(bg_id) + if exitcode_file.exists(): + return f"exited (code {exitcode_file.read_text().strip()})" + try: + os.kill(pid, 0) + except ProcessLookupError: + return "exited (unknown code)" + except PermissionError: + pass + return "running" + + +def _watch_bg_process(bg_id: int, proc: subprocess.Popen) -> None: + """Daemon thread that records the exit code once the child exits.""" + def _wait(): + proc.wait() try: - proc = subprocess.Popen( - command, - shell=True, - cwd=working_dir, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - ) - except Exception as exc: - return f"Error starting background process: {exc}" - bg = BgProcess(id=bg_id, command=command, proc=proc) - _bg_processes[bg_id] = bg - _start_reader_thread(bg) - return f"[bg:{bg_id}] Started: {command} (PID {proc.pid})" - - @tool - def check_process(id: int) -> str: - """Get new output from a background process since the last check. Also reports if it is still running.""" - bg = _bg_processes.get(id) - if bg is None: - return f"Error: no background process with id {id}." - with bg.lock: - new_lines = bg.buffer[bg._read_pos:] - bg._read_pos = len(bg.buffer) - new_output = "".join(new_lines) - status = "running" if bg.proc.poll() is None else f"exited (code {bg.proc.returncode})" - if new_output.strip(): - return f"[bg:{id}] {status}\n{new_output}" - return f"[bg:{id}] {status} (no new output)" - - @tool - def stop_process(id: int) -> str: - """Terminate a background process. Sends SIGTERM, then SIGKILL after 5 seconds.""" - bg = _bg_processes.get(id) - if bg is None: - return f"Error: no background process with id {id}." - if bg.proc.poll() is not None: - return f"[bg:{id}] already exited (code {bg.proc.returncode})" - bg.proc.terminate() + _bg_exitcode_file(bg_id).write_text(str(proc.returncode)) + except OSError: + pass + threading.Thread(target=_wait, daemon=True).start() + + +@tool +def run_background(command: str) -> str: + """Start a long-running process in the background. Returns immediately with a process ID. + Use for servers, file watchers, builds — anything that won't exit quickly.""" + registry = _read_bg_registry() + bg_id = max((int(k) for k in registry), default=0) + 1 + log_file = _bg_log_file(bg_id) + try: + proc = subprocess.Popen( + command, + shell=True, + cwd=_working_dir(), + stdout=open(log_file, "w"), + stderr=subprocess.STDOUT, + ) + except Exception as exc: + return f"Error starting background process: {exc}" + registry[str(bg_id)] = {"command": command, "pid": proc.pid, "log_file": str(log_file)} + _write_bg_registry(registry) + _bg_offset_file(bg_id).write_text("0") + _watch_bg_process(bg_id, proc) + return f"[bg:{bg_id}] Started: {command} (PID {proc.pid})" + + +@tool +def check_process(id: int) -> str: + """Get new output from a background process since the last check. Also reports if it is still running.""" + entry = _read_bg_registry().get(str(id)) + if entry is None: + return f"Error: no background process with id {id}." + log_file = Path(entry["log_file"]) + offset_file = _bg_offset_file(id) + offset = int(offset_file.read_text()) if offset_file.exists() else 0 + new_output = "" + if log_file.exists(): + with open(log_file, "r", errors="replace") as f: + f.seek(offset) + new_output = f.read() + offset_file.write_text(str(f.tell())) + status = _bg_status(id, entry["pid"]) + if new_output.strip(): + return f"[bg:{id}] {status}\n{new_output}" + return f"[bg:{id}] {status} (no new output)" + + +@tool +def stop_process(id: int) -> str: + """Terminate a background process. Sends SIGTERM, then SIGKILL after 5 seconds.""" + entry = _read_bg_registry().get(str(id)) + if entry is None: + return f"Error: no background process with id {id}." + pid = entry["pid"] + status = _bg_status(id, pid) + if status.startswith("exited"): + return f"[bg:{id}] already {status}" + try: + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + return f"[bg:{id}] already exited (unknown code)" + deadline = time.time() + 5 + while time.time() < deadline: + if _bg_status(id, pid) != "running": + break + time.sleep(0.2) + else: try: - bg.proc.wait(timeout=5) - except subprocess.TimeoutExpired: - bg.proc.kill() - bg.proc.wait(timeout=2) - with bg.lock: - final = "".join(bg.buffer[bg._read_pos:]) - bg._read_pos = len(bg.buffer) - status = f"exited (code {bg.proc.returncode})" - if final.strip(): - return f"[bg:{id}] stopped — {status}\n{final}" - return f"[bg:{id}] stopped — {status}" - - @tool - def list_processes() -> str: - """List all background processes with their status.""" - if not _bg_processes: - return "No background processes." - lines = [] - for bg in _bg_processes.values(): - status = "running" if bg.proc.poll() is None else f"exited ({bg.proc.returncode})" - cmd_short = bg.command[:60] + ("..." if len(bg.command) > 60 else "") - lines.append(f" [bg:{bg.id}] PID {bg.proc.pid} {status} {cmd_short}") - return "\n".join(lines) - - def cleanup_all(): - """Kill all background processes. Called on exit.""" - for bg in _bg_processes.values(): - if bg.proc.poll() is None: - bg.proc.terminate() - deadline = time.time() + 5 - for bg in _bg_processes.values(): - remaining = max(0, deadline - time.time()) + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + return f"[bg:{id}] stopped — {_bg_status(id, pid)}" + + +@tool +def list_processes() -> str: + """List all background processes with their status.""" + registry = _read_bg_registry() + if not registry: + return "No background processes." + lines = [] + for bg_id_str, entry in sorted(registry.items(), key=lambda kv: int(kv[0])): + bg_id = int(bg_id_str) + status = _bg_status(bg_id, entry["pid"]) + cmd_short = entry["command"][:60] + ("..." if len(entry["command"]) > 60 else "") + lines.append(f" [bg:{bg_id}] PID {entry['pid']} {status} {cmd_short}") + return "\n".join(lines) + + +def _cleanup_all_bg() -> None: + """Kill all still-running background processes. Called on exit from the main process.""" + registry = _read_bg_registry() + entries = [(int(k), v["pid"]) for k, v in registry.items()] + for bg_id, pid in entries: + if _bg_status(bg_id, pid) == "running": try: - bg.proc.wait(timeout=remaining) - except subprocess.TimeoutExpired: - bg.proc.kill() - - return run_background, check_process, stop_process, list_processes, cleanup_all + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + pass + deadline = time.time() + 5 + while time.time() < deadline and any(_bg_status(i, p) == "running" for i, p in entries): + time.sleep(0.2) + for bg_id, pid in entries: + if _bg_status(bg_id, pid) == "running": + try: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + shutil.rmtree(_bg_state_dir(), ignore_errors=True) # --------------------------------------------------------------------------- @@ -499,158 +564,170 @@ def _consume_events(): # Agent builder # --------------------------------------------------------------------------- +@tool +def read_file(path: str) -> str: + """Read a file and return its text contents. Paths may be absolute or relative to the working directory.""" + working_dir = _working_dir() + target = Path(path) if os.path.isabs(path) else Path(working_dir) / path + if not target.exists(): + return f"Error: {path!r} does not exist." + if target.is_dir(): + return f"Error: {path!r} is a directory. Use list_dir to browse it." + size = target.stat().st_size + if size > _MAX_FILE_BYTES: + return ( + f"Error: {path!r} is {size:,} bytes (limit {_MAX_FILE_BYTES:,}). " + "Use search_in_files to find specific content instead." + ) + try: + return target.read_text(encoding="utf-8", errors="replace") + except Exception as exc: + return f"Error reading {path!r}: {exc}" + + +@tool +def write_file(path: str, content: str) -> str: + """Write content to a file, creating parent directories as needed. Overwrites existing files.""" + working_dir = _working_dir() + target = Path(path) if os.path.isabs(path) else Path(working_dir) / path + try: + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + return f"Wrote {len(content):,} bytes to {str(target)!r}." + except Exception as exc: + return f"Error writing {path!r}: {exc}" + + +@tool +def list_dir(path: str = ".") -> str: + """List directory contents with file sizes. Paths may be absolute or relative to the working directory.""" + working_dir = _working_dir() + target = Path(path) if os.path.isabs(path) else Path(working_dir) / path + if not target.exists(): + return f"Error: {path!r} does not exist." + if not target.is_dir(): + return f"Error: {path!r} is not a directory." + try: + entries = sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name)) + lines = [] + for entry in entries: + if entry.is_dir(): + lines.append(f" {entry.name}/") + else: + lines.append(f" {entry.name} ({entry.stat().st_size:,} bytes)") + header = str(target) + "/" + return header + "\n" + "\n".join(lines) if lines else header + " (empty)" + except Exception as exc: + return f"Error listing {path!r}: {exc}" + + +@tool +def find_files(pattern: str, path: str = ".") -> str: + """Find files matching a glob pattern (e.g. '**/*.py'). Path relative to working directory.""" + working_dir = _working_dir() + base = Path(path) if os.path.isabs(path) else Path(working_dir) / path + if not base.exists(): + return f"Error: {path!r} does not exist." + if not base.is_dir(): + return f"Error: {path!r} is not a directory." + try: + matches = sorted(m for m in base.glob(pattern) if m.is_file()) + if not matches: + return f"No files matching {pattern!r} under {str(base)!r}." + lines = [] + for m in matches[:200]: + try: + rel = m.relative_to(working_dir) + except ValueError: + rel = m + lines.append(str(rel)) + suffix = f"\n... ({len(matches) - 200} more)" if len(matches) > 200 else "" + return "\n".join(lines) + suffix + except Exception as exc: + return f"Error finding files: {exc}" + + +@tool +def search_in_files(regex: str, path: str = ".", file_glob: str = "**/*") -> str: + """Search for a regex pattern in file contents. Returns file:line: matching_line entries.""" + import re as _re + working_dir = _working_dir() + base = Path(path) if os.path.isabs(path) else Path(working_dir) / path + try: + compiled = _re.compile(regex) + except _re.error as exc: + return f"Invalid regex {regex!r}: {exc}" + results = [] + for filepath in sorted(base.glob(file_glob)): + if not filepath.is_file() or filepath.stat().st_size > _MAX_FILE_BYTES: + continue + try: + for lineno, line in enumerate( + filepath.read_text(encoding="utf-8", errors="replace").splitlines(), 1 + ): + if compiled.search(line): + try: + label = str(filepath.relative_to(working_dir)) + except ValueError: + label = str(filepath) + results.append(f"{label}:{lineno}: {line.rstrip()}") + if len(results) >= 100: + break + except Exception: + continue + if len(results) >= 100: + break + if not results: + return f"No matches for {regex!r} in {str(base)!r} ({file_glob})." + suffix = "\n... (truncated at 100 matches)" if len(results) >= 100 else "" + return "\n".join(results) + suffix + + +@tool +def run_shell(command: str) -> str: + """Run a shell command in the working directory. Returns stdout + stderr with exit code. + For long-running commands (servers, watchers), use run_background instead.""" + working_dir = _working_dir() + shell_timeout = _shell_timeout() + try: + proc = subprocess.run( + command, + shell=True, + cwd=working_dir, + capture_output=True, + text=True, + timeout=shell_timeout, + ) + combined = (proc.stdout + proc.stderr).strip() + if len(combined) > _MAX_SHELL_OUTPUT: + combined = combined[:_MAX_SHELL_OUTPUT] + f"\n... (truncated, {len(combined):,} chars total)" + return f"[exit {proc.returncode}]\n{combined}" if combined else f"[exit {proc.returncode}] (no output)" + except subprocess.TimeoutExpired: + return f"Error: command timed out after {shell_timeout}s. Use run_background for long-running commands." + except Exception as exc: + return f"Error: {exc}" + + +@tool +def reply_to_user(message: str) -> str: + """Send your response to the user. Call this when the task is complete.""" + return "ok" + + def build_agent(working_dir: str, shell_timeout: int = _DEFAULT_SHELL_TIMEOUT): """Build the coding agent and return (agent, cleanup_fn). Returns a tuple so the caller can clean up background processes on exit. """ + os.environ[_WORKING_DIR_ENV] = working_dir + os.environ[_SHELL_TIMEOUT_ENV] = str(shell_timeout) + os.environ[_BG_STATE_DIR_ENV] = tempfile.mkdtemp(prefix="coding_agent_tui_bg_") receive_message = wait_for_message_tool( name="wait_for_message", description="Wait for the next user message. Payload has a 'text' field.", ) - # Background process tools (shared registry via closure) - run_background, check_process, stop_process, list_processes, cleanup_bg = ( - _make_bg_tools(working_dir) - ) - - @tool - def read_file(path: str) -> str: - """Read a file and return its text contents. Paths may be absolute or relative to the working directory.""" - target = Path(path) if os.path.isabs(path) else Path(working_dir) / path - if not target.exists(): - return f"Error: {path!r} does not exist." - if target.is_dir(): - return f"Error: {path!r} is a directory. Use list_dir to browse it." - size = target.stat().st_size - if size > _MAX_FILE_BYTES: - return ( - f"Error: {path!r} is {size:,} bytes (limit {_MAX_FILE_BYTES:,}). " - "Use search_in_files to find specific content instead." - ) - try: - return target.read_text(encoding="utf-8", errors="replace") - except Exception as exc: - return f"Error reading {path!r}: {exc}" - - @tool - def write_file(path: str, content: str) -> str: - """Write content to a file, creating parent directories as needed. Overwrites existing files.""" - target = Path(path) if os.path.isabs(path) else Path(working_dir) / path - try: - target.parent.mkdir(parents=True, exist_ok=True) - target.write_text(content, encoding="utf-8") - return f"Wrote {len(content):,} bytes to {str(target)!r}." - except Exception as exc: - return f"Error writing {path!r}: {exc}" - - @tool - def list_dir(path: str = ".") -> str: - """List directory contents with file sizes. Paths may be absolute or relative to the working directory.""" - target = Path(path) if os.path.isabs(path) else Path(working_dir) / path - if not target.exists(): - return f"Error: {path!r} does not exist." - if not target.is_dir(): - return f"Error: {path!r} is not a directory." - try: - entries = sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name)) - lines = [] - for entry in entries: - if entry.is_dir(): - lines.append(f" {entry.name}/") - else: - lines.append(f" {entry.name} ({entry.stat().st_size:,} bytes)") - header = str(target) + "/" - return header + "\n" + "\n".join(lines) if lines else header + " (empty)" - except Exception as exc: - return f"Error listing {path!r}: {exc}" - - @tool - def find_files(pattern: str, path: str = ".") -> str: - """Find files matching a glob pattern (e.g. '**/*.py'). Path relative to working directory.""" - base = Path(path) if os.path.isabs(path) else Path(working_dir) / path - if not base.exists(): - return f"Error: {path!r} does not exist." - if not base.is_dir(): - return f"Error: {path!r} is not a directory." - try: - matches = sorted(m for m in base.glob(pattern) if m.is_file()) - if not matches: - return f"No files matching {pattern!r} under {str(base)!r}." - lines = [] - for m in matches[:200]: - try: - rel = m.relative_to(working_dir) - except ValueError: - rel = m - lines.append(str(rel)) - suffix = f"\n... ({len(matches) - 200} more)" if len(matches) > 200 else "" - return "\n".join(lines) + suffix - except Exception as exc: - return f"Error finding files: {exc}" - - @tool - def search_in_files(regex: str, path: str = ".", file_glob: str = "**/*") -> str: - """Search for a regex pattern in file contents. Returns file:line: matching_line entries.""" - import re as _re - base = Path(path) if os.path.isabs(path) else Path(working_dir) / path - try: - compiled = _re.compile(regex) - except _re.error as exc: - return f"Invalid regex {regex!r}: {exc}" - results = [] - for filepath in sorted(base.glob(file_glob)): - if not filepath.is_file() or filepath.stat().st_size > _MAX_FILE_BYTES: - continue - try: - for lineno, line in enumerate( - filepath.read_text(encoding="utf-8", errors="replace").splitlines(), 1 - ): - if compiled.search(line): - try: - label = str(filepath.relative_to(working_dir)) - except ValueError: - label = str(filepath) - results.append(f"{label}:{lineno}: {line.rstrip()}") - if len(results) >= 100: - break - except Exception: - continue - if len(results) >= 100: - break - if not results: - return f"No matches for {regex!r} in {str(base)!r} ({file_glob})." - suffix = "\n... (truncated at 100 matches)" if len(results) >= 100 else "" - return "\n".join(results) + suffix - - @tool - def run_shell(command: str) -> str: - """Run a shell command in the working directory. Returns stdout + stderr with exit code. - For long-running commands (servers, watchers), use run_background instead.""" - try: - proc = subprocess.run( - command, - shell=True, - cwd=working_dir, - capture_output=True, - text=True, - timeout=shell_timeout, - ) - combined = (proc.stdout + proc.stderr).strip() - if len(combined) > _MAX_SHELL_OUTPUT: - combined = combined[:_MAX_SHELL_OUTPUT] + f"\n... (truncated, {len(combined):,} chars total)" - return f"[exit {proc.returncode}]\n{combined}" if combined else f"[exit {proc.returncode}] (no output)" - except subprocess.TimeoutExpired: - return f"Error: command timed out after {shell_timeout}s. Use run_background for long-running commands." - except Exception as exc: - return f"Error: {exc}" - - @tool - def reply_to_user(message: str) -> str: - """Send your response to the user. Call this when the task is complete.""" - return "ok" - agent = Agent( name="coding_agent_tui", model=settings.llm_model, @@ -704,7 +781,7 @@ def reply_to_user(message: str) -> str: """, ) - return agent, cleanup_bg + return agent, _cleanup_all_bg # --------------------------------------------------------------------------- diff --git a/examples/agents/blog_and_videos/email-subscription-agent/subscription-agent.py b/examples/agents/blog_and_videos/email-subscription-agent/subscription-agent.py index 12fdee10..c5fdc041 100644 --- a/examples/agents/blog_and_videos/email-subscription-agent/subscription-agent.py +++ b/examples/agents/blog_and_videos/email-subscription-agent/subscription-agent.py @@ -324,23 +324,28 @@ def handle_events(handle): print("\n" + summary) -with AgentRuntime() as runtime: - print("\n📬 Hey! I'm your Gmail subscription analyst.") - print("I can find your subscriptions, spot duplicates, flag unused services,") - print("and tell you exactly what to cancel and where to cancel it.") - print("Type 'exit' to quit.\n") - - while True: - try: - prompt = input("You: ").strip() - except (EOFError, KeyboardInterrupt): - print("\nGoodbye! 👋") - break - if not prompt: - continue - if prompt.lower() in ("exit", "quit", "bye"): - print("\nGoodbye! 👋") - break - print() - handle_events(runtime.start(agent, prompt)) - print() \ No newline at end of file +def main() -> None: + with AgentRuntime() as runtime: + print("\n📬 Hey! I'm your Gmail subscription analyst.") + print("I can find your subscriptions, spot duplicates, flag unused services,") + print("and tell you exactly what to cancel and where to cancel it.") + print("Type 'exit' to quit.\n") + + while True: + try: + prompt = input("You: ").strip() + except (EOFError, KeyboardInterrupt): + print("\nGoodbye! 👋") + break + if not prompt: + continue + if prompt.lower() in ("exit", "quit", "bye"): + print("\nGoodbye! 👋") + break + print() + handle_events(runtime.start(agent, prompt)) + print() + + +if __name__ == "__main__": + main() \ No newline at end of file