"""Task execution engine with OpenAI-compatible LLM provider integration. Merged from agent_swarm_v4: - OpenAI-only provider configuration (OPENAI_*/MODEL_* env vars and custom base URLs) - Peer-collaboration hook and specialist-role alignment prompting - Preserves model invocation, handoff decision hooks, workspace summarization, file-application behavior, and the Manager billing/audit usage attribution (X-Agent/X-Agnet headers, usage payload, billing_source) """ import asyncio import json import logging import os import time from pathlib import Path from typing import Callable, Optional from openai import AsyncOpenAI from .handoff_logic import should_handoff logger = logging.getLogger(__name__) class TaskExecutor: """Executes tasks using a configurable OpenAI-compatible API.""" def __init__(self, agent_id: str, workspace_dir: str): self.agent_id = agent_id self.workspace_dir = workspace_dir api_key = ( os.getenv("OPENAI_API_KEY") or os.getenv("MODEL_API_KEY") ) if not api_key: raise ValueError("OPENAI_API_KEY environment variable not set") self.model = ( os.getenv("OPENAI_MODEL") or os.getenv("MODEL_NAME") or os.getenv("MODEL_ID") or "gpt-4o-mini" ) self.api_base = ( os.getenv("OPENAI_API_BASE") or os.getenv("MODEL_API_BASE") or "https://api.openai.com/v1" ) self.api_mode = "openai" self.client = AsyncOpenAI(api_key=api_key, base_url=self.api_base) self.client_type = "openai" self.usage = self._empty_usage() self.current_context: dict = {} # Jina MCP — loaded once at startup; empty list means no tools available self.jina_api_key = os.getenv("JINA_API_KEY", "") self._jina_tools: list[dict] = [] # OpenAI-format tool schemas self._jina_tools_loaded = False async def execute_task( self, task_id: str, description: str, context: dict, handoff_callback: Optional[Callable] = None, agent_capabilities: Optional[list[str]] = None, peer_collaboration_callback: Optional[Callable] = None, proposal_callback: Optional[Callable] = None, ) -> dict: logger.info(f"Executing task {task_id}: {description}") try: started_at = time.time() self.current_context = context or {} self.usage = self._empty_usage(context) # agent_swarm#7 — bottom-up decomposition (the swarm's canonical fan-out path): when this # is a top-level seed, decompose it and PROPOSE the subtasks to the shared pool so peers # self-select them, instead of doing the whole objective solo. Proposed/child tasks never # re-propose (guarded in _maybe_propose_subtasks) → no loop. See autonomous-task-generation.md. proposed = await self._maybe_propose_subtasks(task_id, description, context, proposal_callback) if proposed: return { "success": True, "task_id": task_id, "summary": f"Decomposed the objective into {proposed} subtask(s) and proposed them to the swarm pool for peers to execute (agent_swarm#7).", "subtasks": [], "proposed_subtasks": proposed, "agent_id": self.agent_id, "usage": self._usage_payload(time.time() - started_at), } multi_agent_leaf_mode = self._multi_agent_leaf_mode(context) allow_handoff = self._allow_dynamic_handoff(context) if multi_agent_leaf_mode: subtasks = [{ "description": description, "complexity": "medium", "estimated_time": 30, "dependencies": [], "required_capabilities": context.get("required_capabilities") or ["general"], }] elif self._subtask_handoff_enabled(): subtasks = await self._parse_task(description, context) else: subtasks = [{ "description": description, "complexity": "medium", "estimated_time": 30, "dependencies": [], "required_capabilities": ["general"], }] results = [] for subtask in subtasks: if handoff_callback and self._subtask_handoff_enabled() and allow_handoff: decision = should_handoff( subtask, self.agent_id, agent_capabilities=agent_capabilities, ) if decision.should_handoff: await handoff_callback( task_id=task_id, subtask=subtask, target_capabilities=decision.target_capabilities, ) results.append({ "subtask": subtask, "status": "handed_off", "reason": decision.reason, "target_capabilities": decision.target_capabilities, }) continue results.append(await self._execute_subtask(subtask, task_id, context, peer_collaboration_callback)) success = all(r["status"] in ["completed", "handed_off"] for r in results) awaiting_handoff = any(r["status"] == "handed_off" for r in results) payload = { "success": success, "task_id": task_id, "subtasks": results, "awaiting_handoff": awaiting_handoff, "agent_id": self.agent_id, "usage": self._usage_payload(time.time() - started_at), } # Hoist every subtask's generated files to a top-level `files` array so the produced # artifacts travel in task.result and reach the orchestrator's aggregation node WITHOUT # a git push of a work branch (the swarm's new convergence path; orchestrator # quality.collect_generated_files consumes result['files']). Each entry keeps the frozen # shape {path, content, action} with the COMPLETE file content; last writer wins per path # so a later subtask that rewrites a file supersedes an earlier one. Deletes are carried # through as {path, action:"delete"} (no content) for the aggregator to honor. payload["files"] = self._aggregate_subtask_files(results) if not success: # Surface the model's OWN failure explanation (what it saw / what was missing) as a # top-level `error`, so the orchestrator/Manager records WHY instead of a generic # "Task failed" (agent/main.py uses this as the failure reason). Lead with the model's # summary/error — NOT the task prompt — so the reason reads as the diagnosis, not the # ask. summary is usually the fuller "saw X, missing Y" narrative; append a distinct # error for the concise cause. failed = [r for r in results if r.get("status") not in ("completed", "handed_off")] explanations = [] for r in failed: summary = str(r.get("summary") or "").strip() error = str(r.get("error") or "").strip() text = summary or error or "subtask failed without detail" if summary and error and error not in summary: text = f"{summary} ({error})" explanations.append(text) detail = "; ".join(explanations) or "subtask(s) failed without detail" payload["error"] = detail logger.error(f"Task {task_id} failed: {detail}") return payload except Exception as e: logger.error(f"Error executing task {task_id}: {e}") return { "success": False, "task_id": task_id, "error": str(e), "agent_id": self.agent_id, "usage": self._usage_payload(time.time() - started_at if "started_at" in locals() else 0), } finally: self.current_context = {} @staticmethod def _aggregate_subtask_files(results: list[dict]) -> list[dict]: """Flatten the per-subtask file specs into one ordered, de-duplicated `files` list. Each subtask result carries the model's `files` (the same {path, action, content} specs the executor applied to the workspace). We re-emit them at the top level so the artifacts travel in task.result instead of a pushed git branch. Last writer wins per path (a later subtask rewriting/deleting a file supersedes an earlier write); ordering follows last occurrence. Only writes with non-None content and deletes are kept; malformed entries are skipped. """ by_path: dict[str, dict] = {} for r in results: if not isinstance(r, dict): continue for f in (r.get("files") or []): if not isinstance(f, dict): continue path = f.get("path") if not path: continue action = f.get("action", "write") if action == "delete": by_path[path] = {"path": path, "action": "delete"} elif action == "write": content = f.get("content") if content is None: continue # an incomplete write (no content) is not a usable artifact by_path[path] = {"path": path, "content": content, "action": "write"} # unknown actions are ignored (the orchestrator only consumes write/delete) return list(by_path.values()) async def _maybe_propose_subtasks(self, task_id: str, description: str, context: dict, proposal_callback: Optional[Callable]) -> int: """agent_swarm#7: decompose a top-level seed and propose each subtask to the shared pool. Returns the number of subtasks proposed (0 = not a seed / disabled / undecomposable, caller then executes normally). Guards against loops: only top-level seeds (no parent, source not agent_proposed/dynamic_handoff) propose; the proposed children won't re-propose.""" if not proposal_callback: return 0 ctx = context or {} if ctx.get("parent_task_id") is not None: return 0 if ctx.get("source") in ("agent_proposed", "dynamic_handoff"): return 0 if os.getenv("ENABLE_AUTONOMOUS_PROPOSALS", "true").lower() not in {"1", "true", "yes"}: return 0 try: subtasks = await self._parse_task(description, ctx) except Exception as e: logger.warning(f"Seed decomposition failed: {e}") return 0 if not subtasks or len(subtasks) < 2: return 0 # nothing to fan out — execute as a single task proposed = 0 for st in subtasks: desc = (st.get("description") or "").strip() if not desc: continue caps = st.get("required_capabilities") or [] try: await proposal_callback( description=desc, reason="bottom-up decomposition of the seed objective (agent_swarm#7)", confidence=0.8, origin_task_id=task_id, trigger_event="seed_decomposition", shared_state_snapshot={"origin_task": task_id}, # Capabilities are a HINT (agent_role), not a dispatch gate: leave # required_capabilities empty so ANY idle agent can self-select the subtask # (like the seed). LLM-specific caps (e.g. flask/sqlite) aren't a subset of the # fixed pool caps → can_agent_run_task would reject → task stuck PENDING (the # dispatch gap we hit). Self-selection + τ handle routing instead. agent_role=",".join(caps) or "general", required_capabilities=[], ) proposed += 1 except Exception as e: logger.warning(f"propose_task failed for a subtask: {e}") if proposed: logger.info(f"Proposed {proposed} subtask(s) to the shared pool (agent_swarm#7)") return proposed async def _parse_task(self, description: str, context: dict) -> list[dict]: # agent_swarm#75 (LOCAL fix — not for upstream as-is): the old impl capped output at # max_tokens=2000 + strict json.loads + silently fell back to a SINGLE task on any error, # so large tasks (their breakdown JSON exceeds 2000 → truncated → parse fail) never split → # the swarm degraded to one agent. Fix: bigger configurable budget, lenient parsing # (accept bare array or {"subtasks":[...]}), one retry, and explicit decompose/fallback logs. # Decomposition output budget (env-tunable). qwen3.7-max max output = 65536 tokens # (gateway rejects >65536 with HTTP 400 InvalidParameter), so default to the full 65536 # so large-task breakdown JSON never truncates. It's a cap, not a target. max_tokens = int(os.getenv("AGENT_PLAN_MAX_TOKENS", "65536") or 65536) prompt = f"""You are decomposing a software change for a DECENTRALIZED SWARM of autonomous coding agents. How the swarm executes (this CONSTRAINS what a valid subtask is — read carefully): - The target repository is ALREADY checked out locally at the correct base commit. Each agent can READ any file in the repo and its ONLY output is the COMPLETE new content of the file(s) it owns. - Subtasks go into a shared pool and are picked up by DIFFERENT agents IN PARALLEL. Agents do NOT share a filesystem and there is NO guaranteed order — every subtask must be independently doable straight from the base commit, without seeing any other agent's output. - All agents' file outputs are merged BY PATH (one file = one owner) and diffed against the base commit to produce the final patch. So every subtask MUST be a concrete CODE CHANGE scoped to specific file(s) that yields complete file content. A subtask must NOT be a process/meta step. FORBIDDEN subtasks (an agent literally cannot do these — they will fail and waste the run): exploring/surveying the repo, running or executing tests, moving/renaming/deleting files as an \"action\", reviewing or verifying other agents' work, setting up the environment, installing dependencies, committing/pushing, \"ensure/confirm/check that ...\". ALLOWED subtask shape: \"Implement/modify in , outputting the complete updated file.\" Writing a NEW source or test FILE is allowed (it produces content); \"run the tests\" is not. Rules: 1. Partition by FILE OWNERSHIP: each file that must change is owned by EXACTLY ONE subtask. Never let two subtasks edit the same file (parallel agents would clobber each other). 2. One subtask may own several files only if they must change together; keep ownership disjoint. 3. Make the SMALLEST change that satisfies the task. If it fits in one file, return EXACTLY ONE subtask — do not invent extra steps to look thorough. 4. Ground every subtask in the ACTUAL files present at the base commit, not assumptions. 5. dependencies = real content dependencies only (e.g. a test file depends on the intended interface of an impl file); each subtask must still be writable independently from the base commit. Task: {description} Context: {json.dumps(context, indent=2)} Return ONLY a JSON array. Each element: - description: imperative and file-scoped — WHAT to change, in WHICH file(s), and that it must output the complete final file content. - target_files: array of repo-relative paths this subtask OWNS (disjoint across subtasks). - complexity: \"low\" | \"medium\" | \"high\" - dependencies: array of subtask indices (content dependencies only; empty if none) - required_capabilities: e.g. [\"python\"], [\"python\",\"testing\"] Return ONLY the JSON array, no other text.""" last_err = None for attempt in range(2): # one retry before degrading try: content = await self._complete(prompt, max_tokens=max_tokens) subtasks = self._coerce_subtasks(content) if subtasks: logger.info(f"Decomposed task into {len(subtasks)} subtask(s)") return subtasks last_err = "no subtasks parsed" except Exception as e: last_err = e logger.warning(f"Task decomposition parse failed (attempt {attempt + 1}/2): {e}") logger.warning(f"Task decomposition fell back to a single task (reason: {last_err}); the run will not fan out") return [{ "description": description, "complexity": "medium", "estimated_time": 30, "dependencies": [], "required_capabilities": ["general"], }] def _coerce_subtasks(self, content: str) -> list[dict]: """Lenient parse of the planner output into a subtask list (agent_swarm#75). Accepts a bare JSON array, a ```-fenced array, or an object like {"subtasks": [...]}. Returns [] when nothing usable is found (caller retries / degrades).""" text = self._strip_json_fence(content or "") obj = None try: obj = json.loads(text) except Exception: start, end = text.find("["), text.rfind("]") if start != -1 and end > start: try: obj = json.loads(text[start:end + 1]) except Exception: obj = None if isinstance(obj, dict): obj = obj.get("subtasks") or obj.get("tasks") if not isinstance(obj, list): return [] return [s for s in obj if isinstance(s, dict) and s.get("description")] async def _execute_subtask(self, subtask: dict, task_id: str, context: dict, peer_collaboration_callback: Optional[Callable]) -> dict: content = "" try: description = subtask["description"] logger.info("════════ [任务开始] task=%s role=%s\n 描述: %s", task_id, context.get("specialist_role", "general"), description) workspace_files = self._summarize_workspace() workspace_context = self._collect_workspace_context() user_prompt = context.get("user_prompt") or context.get("run_goal") or context.get("root_task_description") or "" specialist_role = context.get("specialist_role", "general") dependency_artifacts = context.get("dependency_artifacts") or [] implementation_artifacts = [ artifact for artifact in dependency_artifacts if "implementation" in (artifact.get("task_id") or "") ] testing_artifacts = [ artifact for artifact in dependency_artifacts if "testing" in (artifact.get("task_id") or "") ] peer_context = [] peer_agents = context.get("peer_agents") or [] should_consult_peers = specialist_role in {"testing", "documentation"} if peer_collaboration_callback and peer_agents and should_consult_peers: max_peer_consults = int((context or {}).get("max_peer_consults", 2) or 2) preferred_roles = [] if specialist_role in {"testing", "documentation"}: preferred_roles = ["implementation"] ordered_peers = sorted( peer_agents, key=lambda peer: 0 if peer.get("role") in preferred_roles else 1, ) for peer in ordered_peers[:max_peer_consults]: try: reply = await peer_collaboration_callback( task_id=task_id, target_agent_id=peer.get("agent_id"), content=( f"Specialist role: {specialist_role}. " f"Please provide guidance and confirm behavior for: {description}. " f"Original user request: {user_prompt}" ), timeout_seconds=float((context or {}).get("peer_timeout_seconds", 10.0) or 10.0), ) peer_context.append( { "agent_id": peer.get("agent_id"), "role": peer.get("role"), "content": reply.get("content", ""), } ) except Exception as exc: logger.warning(f"Peer collaboration failed with {peer.get('agent_id')}: {exc}") prompt = f"""You are a programming assistant working in a collaborative agent system. Original user request: {user_prompt} Your specialist role: {specialist_role} Task: {description} Dependency artifacts from other specialists: {json.dumps(dependency_artifacts, indent=2)} Implementation artifacts relevant to alignment: {json.dumps(implementation_artifacts, indent=2)} Testing artifacts relevant to alignment: {json.dumps(testing_artifacts, indent=2)} Workspace directory: {self.workspace_dir} Current workspace files: {json.dumps(workspace_files, indent=2)} Relevant workspace file contents: {json.dumps(workspace_context, indent=2)} Peer specialist input: {json.dumps(peer_context, indent=2)} Workspace & tool boundaries (IMPORTANT): - The COMPLETE target repository is ALREADY cloned on local disk at: {self.workspace_dir} (checked out at the correct base commit). The file list and contents shown above are only a PARTIAL SAMPLE of that repo — NOT the whole tree. So NEVER conclude a file or directory is "missing" just because it is not listed above: if the task references e.g. `qutebrowser/utils/log.py`, that file EXISTS on disk at {self.workspace_dir}/qutebrowser/utils/log.py — assume it is present and edit it. - Treat the local files as the single source of truth. Return the complete modified content of each file you change in `files` (full file, not a diff). - The web search / read_url tools are for EXTERNAL knowledge ONLY (library/framework docs, language features, error-message lookups). NEVER use them to fetch THIS repository's own source — the web copy is a DIFFERENT version than your local base commit and will corrupt your patch. - Do not pass local paths or file:// URLs to read_url; it only fetches public web URLs. Alignment requirements: - If your role is testing, align your tests with the implementation artifacts and their stated error semantics. - If your role is documentation, align your docs with both implementation and testing artifacts. - Do not invent behavior that conflicts with dependency artifacts unless you explicitly surface an error. - Treat implementation artifacts as the source of truth for API behavior and exception semantics. - If peer specialist input conflicts with implementation artifacts, prefer implementation semantics and explain the correction in your changes summary. - If your role is documentation or testing, update only your specialist outputs to converge on implementation behavior unless the implementation artifact is clearly missing or contradictory. Return your response as JSON with this structure: {{ \"status\": \"completed\" or \"failed\", \"summary\": \"Brief description of what was done\", \"files\": [{{\"path\": \"relative/path.py\", \"action\": \"write\", \"content\": \"complete file content\"}}], \"changes\": \"Detailed description of changes\", \"error\": \"Error message if failed, null otherwise\" }} Return ONLY the JSON, no other text.""" # Full-file rewrites (the `files` JSON carries complete file contents) easily exceed a few # thousand tokens → 4000 truncated the JSON mid-string → parse failure → task failed. # Raise to the model's output ceiling (qwen3.7-max = 65536); env-tunable for other gateways. exec_max_tokens = int(os.getenv("AGENT_EXEC_MAX_TOKENS", "65535") or 65535) content = await self._complete(prompt, max_tokens=exec_max_tokens) result = self._parse_json_response(content) logger.info("──────── [任务产出] task=%s status=%s 文件=%s\n 说明: %s", task_id, result.get("status"), [f.get("path") for f in (result.get("files") or [])], (result.get("changes") or result.get("summary") or "")[:600]) if result.get("status") == "completed": apply_result = await self._apply_file_changes(result.get("files", [])) result["files_modified"] = apply_result["files_modified"] result["files_deleted"] = apply_result["files_deleted"] if apply_result["errors"]: result["status"] = "failed" result["error"] = "; ".join(apply_result["errors"]) logger.warning( f"Subtask for task {task_id} failed applying file changes: {result['error']}" ) else: # LLM returned 2xx but declared the subtask not completed: log its reason and a # bounded snippet of the model output so the failure is diagnosable from agent logs. logger.warning( "Subtask for task %s reported status=%r (error=%s summary=%s); llm_response[:500]=%s", task_id, result.get("status"), result.get("error"), result.get("summary"), (content or "").strip()[:500], ) result["subtask"] = subtask return result except Exception as e: # Parse failures / API errors land here; include a bounded model-output snippet # (model-generated text, no secrets) to explain why parsing/execution failed. logger.error( f"Error executing subtask for task {task_id}: {e}; " f"llm_response[:500]={(content or '').strip()[:500]}" ) return { "subtask": subtask, "status": "failed", "error": str(e), "summary": f"Failed to execute: {e}", } # Jina MCP endpoint (StreamableHTTP). Read-only web tools (search_web/read_url/…). _JINA_MCP_URL = "https://mcp.jina.ai/v1" def _jina_mcp_headers(self) -> dict: return {"Authorization": f"Bearer {self.jina_api_key}"} async def _load_jina_tools(self) -> list[dict]: """Fetch tool schemas from Jina MCP via the standard `mcp` SDK (StreamableHTTP transport). Cached after first call. Returns OpenAI function-calling tool specs. The SDK handles the MCP handshake, SSE framing, and session — no hand-rolled JSON-RPC/SSE parsing.""" if self._jina_tools_loaded: return self._jina_tools self._jina_tools_loaded = True if not self.jina_api_key: return [] try: from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async with streamablehttp_client( self._JINA_MCP_URL, headers=self._jina_mcp_headers() ) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() tools = (await session.list_tools()).tools # Constrain the toolset: the agent only needs web SEARCH + fetch a page's full # CONTENT. Exposing all ~21 Jina tools bloats the prompt and tempts the model to # misuse them (e.g. read_url on local file:// paths, or fetching the repo's own # source from the web = WRONG version vs the local base_commit checkout). allow = {t.strip() for t in os.getenv( "JINA_TOOL_ALLOWLIST", "search_web,read_url").split(",") if t.strip()} self._jina_tools = [ { "type": "function", "function": { "name": t.name, "description": t.description or "", "parameters": t.inputSchema or {"type": "object", "properties": {}}, }, } for t in tools if not allow or t.name in allow ] logger.info("Loaded %d Jina MCP tools (allowlist=%s, of %d offered)", len(self._jina_tools), sorted(allow) or "*", len(tools)) except Exception as exc: logger.warning("Failed to load Jina MCP tools: %s", exc) self._jina_tools = [] return self._jina_tools async def _call_jina_tool(self, tool_name: str, arguments: dict) -> str: """Invoke a single Jina MCP tool via the standard `mcp` SDK and return its text result.""" try: from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async with streamablehttp_client( self._JINA_MCP_URL, headers=self._jina_mcp_headers() ) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() result = await session.call_tool(tool_name, arguments) parts = [ c.text for c in result.content if getattr(c, "type", None) == "text" ] return "\n".join(parts) or str(result.content) except Exception as exc: return f"[tool error: {exc}]" @staticmethod def _message_text(msg) -> str: """Robustly extract the assistant's textual answer. qwen3.7-max (a THINKING model) can return an empty `content` with the real answer in `reasoning_content`; reading only `content` then yields "" → `json.loads("")` → "Expecting value: line 1 column 1 (char 0)" → task fails. Fallback chain: content → reasoning_content (attribute or OpenAI-SDK model_extra).""" text = (getattr(msg, "content", None) or "").strip() if text: return text rc = getattr(msg, "reasoning_content", None) if not rc: extra = getattr(msg, "model_extra", None) or {} rc = extra.get("reasoning_content") if isinstance(extra, dict) else None return (rc or "").strip() async def _complete(self, prompt: str, max_tokens: int) -> str: """Call the LLM with optional Jina MCP tools; handles the tool-call loop.""" extra_headers = self._model_attribution_headers() tools = await self._load_jina_tools() messages = [{"role": "user", "content": prompt}] for round_i in range(8): # max 8 tool-call rounds kwargs: dict = dict( model=self.model, messages=messages, max_tokens=max_tokens, extra_headers=extra_headers or None, ) if tools: kwargs["tools"] = tools kwargs["tool_choice"] = "auto" response = await self.client.chat.completions.create(**kwargs) self._record_openai_usage(response) msg = response.choices[0].message # 记录这一轮 LLM 的"思考/回复"(可观测 agent 怎么想的) if msg.content: logger.info("[LLM·第%d轮] 思考/回复:\n%s", round_i + 1, msg.content.strip()[:1200]) if not msg.tool_calls: return self._message_text(msg) # Execute each tool call and feed results back messages.append(msg.model_dump(exclude_unset=True)) for tc in msg.tool_calls: args = json.loads(tc.function.arguments or "{}") # 记录调用了哪个 tool、query 是什么 logger.info("[TOOL·调用] %s 参数=%s", tc.function.name, json.dumps(args, ensure_ascii=False)[:400]) result = await self._call_jina_tool(tc.function.name, args) # 记录 tool 返回了什么(搜索结果内容,截断) logger.info("[TOOL·返回] %s (%d字):\n%s", tc.function.name, len(result), (result or "").strip()[:1000]) messages.append({ "role": "tool", "tool_call_id": tc.id, "content": result, }) # Fallback: ask for a final answer without tools messages.append({"role": "user", "content": "Please provide your final answer now."}) response = await self.client.chat.completions.create( model=self.model, messages=messages, max_tokens=max_tokens, extra_headers=extra_headers or None, ) self._record_openai_usage(response) return self._message_text(response.choices[0].message) async def peer_reply(self, *, query: str, capabilities: list[str], last_summary: Optional[str], max_tokens: Optional[int] = None) -> dict: """Compose a substantive, grounded reply to a peer agent's query (one bounded LLM call). Returns {stance, content, evidence, refs}. `content` is what the requesting agent reads. Bounded by PEER_CONSULT_MAX_TOKENS (default 500). """ if max_tokens is None: max_tokens = int(os.getenv("PEER_CONSULT_MAX_TOKENS", "500")) workspace_files = self._summarize_workspace() workspace_context = self._collect_workspace_context(max_files=8) prompt = f"""You are a specialist agent being consulted by a peer in a collaborative swarm. Your capabilities: {', '.join(capabilities)} Your latest completed work (summary): {last_summary or 'none'} A peer asks: {query} Your workspace files: {json.dumps(workspace_files, indent=2)} Relevant workspace file contents: {json.dumps(workspace_context, indent=2)} Answer concisely and concretely, grounded in YOUR actual work/artifacts. Treat implementation artifacts as the source of truth for behavior and exception semantics; if the peer's assumption conflicts with your work, say so. Return ONLY JSON: {{\"stance\": \"agree|disagree|info\", \"content\": \"\", \"evidence\": \"\", \"refs\": [\"relative/file/path\"]}}""" content = await self._complete(prompt, max_tokens=max_tokens) result = self._parse_json_response(content) if not result.get("content"): result["content"] = (content or "").strip()[:1000] result.setdefault("stance", "info") result.setdefault("evidence", "") result.setdefault("refs", []) return result def _empty_usage(self, context: Optional[dict] = None) -> dict: plan = ((context or {}).get("orchestration_plan") or {}) billing = plan.get("billing_context") or {} return { "model_id": self.model, "model_tokens": 0, "prompt_tokens": 0, "completion_tokens": 0, "model_cost_usd": 0.0, "runtime_seconds": 0.0, "billing_source": billing.get("provider") or os.getenv("BILLING_SOURCE", "unknown"), } def _record_openai_usage(self, response): usage = getattr(response, "usage", None) if not usage: return prompt_tokens = int(getattr(usage, "prompt_tokens", 0) or 0) completion_tokens = int(getattr(usage, "completion_tokens", 0) or 0) total_tokens = int(getattr(usage, "total_tokens", 0) or prompt_tokens + completion_tokens) self._add_usage(prompt_tokens, completion_tokens, total_tokens) def _add_usage(self, prompt_tokens: int, completion_tokens: int, total_tokens: int): self.usage["prompt_tokens"] += prompt_tokens self.usage["completion_tokens"] += completion_tokens self.usage["model_tokens"] += total_tokens input_cost = float(os.getenv("MODEL_INPUT_COST_PER_1M", "0") or 0) output_cost = float(os.getenv("MODEL_OUTPUT_COST_PER_1M", "0") or 0) self.usage["model_cost_usd"] += ( prompt_tokens * input_cost / 1_000_000 + completion_tokens * output_cost / 1_000_000 ) def _usage_payload(self, runtime_seconds: float) -> dict: usage = dict(self.usage) usage["runtime_seconds"] = runtime_seconds return usage def _model_attribution_headers(self) -> dict: headers = {} mapping = { "manager_deployment_id": ["X-Agent-Manager-Deployment-ID", "X-Agnet-Manager-Deployment-ID"], "swarm_id": ["X-Agent-Swarm-ID", "X-Agnet-Swarm-ID"], "task_id": ["X-Agent-Task-ID", "X-Agnet-Task-ID"], "agent_role": ["X-Agent-Agent-Role", "X-Agnet-Agent-Role"], "correlation_id": ["X-Correlation-ID"], "model_id": ["X-Agent-Model-ID", "X-Agnet-Model-ID"], } for key, header_names in mapping.items(): value = self.current_context.get(key) if value: for header in header_names: headers[header] = str(value) headers.setdefault("X-Agent-Model-ID", self.model) headers.setdefault("X-Agnet-Model-ID", self.model) return headers def _subtask_handoff_enabled(self) -> bool: return os.getenv("ENABLE_SUBTASK_HANDOFF", "false").lower() in {"1", "true", "yes"} def _multi_agent_leaf_mode(self, context: Optional[dict]) -> bool: return self._subtask_handoff_enabled() and (context or {}).get("workflow_mode") == "multi_agent" def _allow_dynamic_handoff(self, context: Optional[dict]) -> bool: if (context or {}).get("workflow_mode") != "multi_agent": return True return bool((context or {}).get("allow_handoff", False)) def _strip_json_fence(self, content: str) -> str: content = content.strip() if not content.startswith("```"): return content lines = content.split("\n") if lines and lines[0].startswith("```"): lines = lines[1:] if lines and lines[-1].strip() == "```": lines = lines[:-1] return "\n".join(lines).strip() def _parse_json_response(self, content: str) -> dict: content = self._strip_json_fence(content) try: return json.loads(content) except json.JSONDecodeError: start = content.find("{") end = content.rfind("}") if start == -1 or end == -1 or end <= start: raise return json.loads(content[start:end + 1]) def _summarize_workspace(self, max_files: int = 80) -> list[str]: root = Path(self.workspace_dir) if not root.exists(): return [] ignored_dirs = {".git", "__pycache__", "node_modules", ".venv", "venv"} files = [] for path in root.rglob("*"): if len(files) >= max_files: break if not path.is_file(): continue if any(part in ignored_dirs for part in path.relative_to(root).parts): continue files.append(str(path.relative_to(root))) return sorted(files) def _collect_workspace_context(self, max_files: int = 20, max_bytes_per_file: int = 6000) -> dict[str, str]: # Real-repo edits need the FULL file (the LLM rewrites complete content; truncated input → # truncated/hallucinated rewrite → wrong diff). Raise budgets, env-tunable for big repos. max_files = int(os.getenv("AGENT_CTX_MAX_FILES", str(max_files))) max_bytes_per_file = int(os.getenv("AGENT_CTX_MAX_BYTES", str(max_bytes_per_file))) root = Path(self.workspace_dir) if not root.exists(): return {} ignored_dirs = {".git", "__pycache__", "node_modules", ".venv", "venv"} allowed_suffixes = {".py", ".js", ".ts", ".tsx", ".jsx", ".json", ".md", ".txt", ".yaml", ".yml", ".toml", ".ini", ".cfg", ".sh"} context = {} for path in sorted(root.rglob("*")): if len(context) >= max_files: break if not path.is_file(): continue relative = path.relative_to(root) if any(part in ignored_dirs for part in relative.parts): continue if path.suffix and path.suffix.lower() not in allowed_suffixes: continue try: data = path.read_bytes()[:max_bytes_per_file] context[str(relative)] = data.decode("utf-8") except UnicodeDecodeError: continue except Exception as e: logger.warning(f"Failed to read workspace file {relative}: {e}") return context def _resolve_workspace_path(self, file_path: str) -> Path: if not file_path or os.path.isabs(file_path): raise ValueError(f"Invalid relative path: {file_path}") root = Path(self.workspace_dir).resolve() resolved = (root / file_path).resolve() if root != resolved and root not in resolved.parents: raise ValueError(f"Path escapes workspace: {file_path}") return resolved async def _apply_file_changes(self, files: list[dict]) -> dict: result = {"files_modified": [], "files_deleted": [], "errors": []} for file_change in files: path = file_change.get("path") action = file_change.get("action", "write") try: target = self._resolve_workspace_path(path) if action == "write": content = file_change.get("content") if content is None: raise ValueError(f"Missing content for {path}") target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") result["files_modified"].append(path) elif action == "delete": if target.exists(): target.unlink() result["files_deleted"].append(path) else: raise ValueError(f"Unsupported file action for {path}: {action}") except Exception as e: logger.error(f"Failed to apply file change {path}: {e}") result["errors"].append(f"{path}: {e}") return result async def read_file(self, file_path: str) -> Optional[str]: try: full_path = os.path.join(self.workspace_dir, file_path) with open(full_path, "r", encoding="utf-8") as f: return f.read() except Exception as e: logger.error(f"Error reading file {file_path}: {e}") return None async def write_file(self, file_path: str, content: str) -> bool: try: full_path = self._resolve_workspace_path(file_path) os.makedirs(os.path.dirname(full_path), exist_ok=True) with open(full_path, "w", encoding="utf-8") as f: f.write(content) logger.info(f"Wrote file: {file_path}") return True except Exception as e: logger.error(f"Error writing file {file_path}: {e}") return False