"""Runs discussions and hands the settled ones to the pipeline.""" from __future__ import annotations from pathlib import Path from pydantic import ValidationError from .adapters.parse import classify_failure from .discussion import ( Brief, Discussion, DiscussionRecord, DiscussionSpec, new_id, title_from, validate_id, ) from .domain import TaskRecord, TaskSpec class DiscussionNotFound(FileNotFoundError): pass class AdvisorUnavailable(RuntimeError): """The provider could not answer. Nothing was lost; the discussion is intact.""" def _explain(error: str | None) -> str: """Say what actually went wrong instead of pasting the provider's stack of errors. A wall of `401 Unauthorized` websocket lines tells an operator nothing about the fact that the account running the service simply never logged in. """ text = (error or "").strip() if not text: return "没有回应" category = classify_failure(text) head = _telling_line(text, category) if category == "AUTH": return ( "provider 未登录。运行本服务的系统账号需要自己登录一次" "（凭据存在该账号的家目录里，不会从别的用户继承）。原始信息：" + head ) if category == "RATE_LIMIT": return "provider 限流，稍后再试。原始信息：" + head if category == "NETWORK": return "连不上 provider，检查网络或代理设置。原始信息：" + head return head def _telling_line(text: str, category: str) -> str: """The line that explains the failure, not merely the first one printed. Providers open with notices like "Reading additional input from stdin"; quoting that back would hide the actual cause sitting three lines below it. """ lines = [line.strip() for line in text.splitlines() if line.strip()] for line in lines: if classify_failure(line) == category: return line[:160] return lines[0][:160] if lines else "" def _extract_json(text: str) -> str: """Pull the JSON object out of a prose reply that may be wrapped in a fence.""" body = text.strip() if body.startswith("```"): body = body.split("```")[1] if "```" in body[3:] else body[3:] body = body[4:] if body.lower().startswith("json") else body start, end = body.find("{"), body.rfind("}") return body[start : end + 1] if start != -1 and end > start else body.strip() class Advisor: """Free-form discussion with the agents, before any task exists. Nothing here touches a worktree or advances a state machine, which is why prose is allowed. The only way out of a discussion is a validated `Brief`, so what the pipeline receives is structured data even though the conversation that produced it was not. """ def __init__(self, store, planner, implementer, orchestrator=None) -> None: self.store = store self.planner = planner # Codex by default: the one that plans self.implementer = implementer # Claude: the second opinion self.orchestrator = orchestrator # ------------------------------------------------------------------ lookup def _open(self, discussion_id: str) -> Discussion: discussion = Discussion(self.store.root, validate_id(discussion_id)) if not discussion.exists(): raise DiscussionNotFound(discussion_id) return discussion def list(self) -> list[DiscussionRecord]: root = self.store.root / "discussions" if not root.is_dir(): return [] found = [] for entry in sorted(root.iterdir()): try: found.append(Discussion(self.store.root, entry.name).record()) except Exception: # noqa: BLE001 - one unreadable discussion must not hide the rest continue return sorted(found, key=lambda record: record.updated_at, reverse=True) def record(self, discussion_id: str) -> DiscussionRecord: return self._open(discussion_id).record() def messages(self, discussion_id: str) -> list: return self._open(discussion_id).messages() # ------------------------------------------------------------------ talking def start(self, spec: DiscussionSpec) -> DiscussionRecord: repo = Path(spec.repo_path).resolve(strict=False) if self.orchestrator is not None: repo = self.orchestrator._check_repo(repo) self.orchestrator._validate_tuning(spec.tuning) discussion = Discussion(self.store.root, new_id()) record = discussion.save( DiscussionRecord( discussion_id=discussion.discussion_id, repo_path=repo, title=title_from(spec.opening), tuning=spec.tuning, ) ) discussion.add("operator", spec.opening) return record def say(self, discussion_id: str, text: str, voices: list[str] | None = None) -> list: """Record what the user said and collect the requested replies.""" if not text.strip(): raise ValueError("an empty message carries nothing") discussion = self._open(discussion_id) discussion.add("operator", text) return self.reply(discussion_id, voices) def reply(self, discussion_id: str, voices: list[str] | None = None) -> list: """Ask the named agents to respond to the discussion so far.""" discussion = self._open(discussion_id) record = discussion.record() wanted = voices or ["codex"] answers = [] failures = [] for voice in wanted: adapter = self._adapter(record, voice) ask = getattr(adapter, "ask", None) if ask is None: failures.append(f"{voice} 不支持自由问答") continue reply = ask( discussion.prompt(record.repo_path), record.repo_path, record.session_ids.get(voice), ) if not reply.ok or not reply.text.strip(): failures.append(f"{voice}：{_explain(reply.error)}") continue answers.append(discussion.add(voice, reply.text)) record = self._remember(discussion, record, voice, reply) if not answers: raise AdvisorUnavailable("；".join(failures) or "没有可用的顾问") return answers def _adapter(self, record: DiscussionRecord, voice: str): """The adapter for this voice, honouring a model or effort chosen for the discussion.""" role = "planner" if voice == "codex" else "implementer" default = self.planner if voice == "codex" else self.implementer wanted = record.tuning.get(role) factory = getattr(self.orchestrator, "adapter_factory", None) if not wanted or not (wanted.model or wanted.effort) or factory is None: return default try: return factory(role, wanted) except Exception: # noqa: BLE001 - a bad choice must not end the discussion return default def _remember(self, discussion: Discussion, record: DiscussionRecord, voice: str, reply) -> DiscussionRecord: updates = {} if reply.session_id: updates["session_ids"] = {**record.session_ids, voice: reply.session_id} if reply.usage: total = dict(record.usage) for key, value in reply.usage.items(): if isinstance(value, (int, float)): total[key] = round(total.get(key, 0) + value, 6) updates["usage"] = total return discussion.save(record.model_copy(update=updates)) if updates else record # ------------------------------------------------------------- handing over def draft_brief(self, discussion_id: str) -> Brief: """Ask the planner to turn the discussion into a requirement you can check.""" discussion = self._open(discussion_id) record = discussion.record() ask = getattr(self._adapter(record, "codex"), "ask", None) if ask is None: raise AdvisorUnavailable("规划方不支持自由问答，无法总结") reply = ask( discussion.brief_prompt(record.repo_path), record.repo_path, record.session_ids.get("codex"), ) if not reply.ok: raise AdvisorUnavailable(_explain(reply.error)) try: brief = Brief.model_validate_json(_extract_json(reply.text)) except (ValidationError, ValueError) as error: raise AdvisorUnavailable(f"总结不是合法的需求结构：{error}") from error discussion.save(record.model_copy(update={"brief": brief})) discussion.add("codex", "已整理出需求草案，请确认。") return brief def set_brief(self, discussion_id: str, brief: Brief) -> DiscussionRecord: """Accept an edited brief. The operator always has the last word on it.""" discussion = self._open(discussion_id) return discussion.save(discussion.record().model_copy(update={"brief": brief})) def to_task(self, discussion_id: str) -> TaskRecord: """Create the task. The pipeline only ever sees the validated brief.""" if self.orchestrator is None: raise AdvisorUnavailable("没有可用的编排器") discussion = self._open(discussion_id) record = discussion.record() if record.brief is None: raise AdvisorUnavailable("还没有定稿的需求，先点「整理需求」") if record.task_id: raise AdvisorUnavailable(f"这个讨论已经建过任务 {record.task_id}") task = self.orchestrator.create_task( TaskSpec( repo_path=record.repo_path, goal=record.brief.goal, constraints=record.brief.constraints, tuning=record.tuning, ) ) discussion.save(record.model_copy(update={"task_id": task.task_id})) self._seed_task_conversation(task.task_id, discussion, record.brief) return task def _seed_task_conversation(self, task_id: str, discussion: Discussion, brief: Brief) -> None: """Carry the agreed requirement into the task so the planner starts from it.""" talk = self.orchestrator.conversation(task_id) talk.add("orchestrator", f"由讨论 {discussion.discussion_id} 转入。", stage="INIT") lines = [f"讨论确定的目标：{brief.goal}"] if brief.constraints: lines.append("约束：" + "；".join(brief.constraints)) if brief.acceptance_criteria: lines.append("验收标准：" + "；".join(brief.acceptance_criteria)) if brief.open_questions: lines.append("仍未确定：" + "；".join(brief.open_questions)) talk.add("operator", "\n".join(lines), stage="INIT")