"""对话目标 → 分阶段页面执行 → 来源分析 → 独立复盘。""" from __future__ import annotations import asyncio import json import re import uuid from copy import deepcopy from typing import Any from backend.browser import runner from backend.browser.llm_factory import get_llm_config from backend.config import settings from backend.db import database as db from backend.orchestrator.progress_doc import append_progress, init_progress_doc, task_doc_dir, write_summary from backend.prospecting.analysis import analyze_sources from backend.prospecting.contracts import ALLOWED_DOMAINS, BUSINESS_ID, KNOWLEDGE_ID, ProspectingBrief from backend.prospecting.dialogue import interpret_message from backend.prospecting.prompts import browser_prompt from backend.prospecting.request_routing import browser_recheck_requested, browser_recheck_brief, ground_execution_plan, recheck_without_analysis from backend.browser.preflight_policy import PreflightPolicy, compile_preflight_policy BRIEF_FIELDS = set(ProspectingBrief.model_fields) EXECUTION_POLICY = "conversation-v1" REQUIREMENTS_GROUNDING_VERSION = "request-fields-v2" CONTROL_CONTINUE = {"继续", "继续执行", "已处理，请继续", "已登录，继续", "用户已处理，请继续"} CONTROL_STOP = {"停止", "暂停", "停一下", "暂停一下", "停止任务", "停止执行", "先停下来", "先停一下"} CONTROL_STATUS = {"进度怎么样", "进度怎么样了", "进展怎么样", "现在进度怎么样", "当前进度", "查看进度", "任务进度", "现在什么情况", "做到哪了", "你现在在干嘛"} def ranked_experience(task: dict) -> list[dict]: from backend.prospecting import knowledge from backend.prospecting.verification import context_for context = context_for(task) available = knowledge.list_methods(context, include_legacy=False) context["available_method_ids"] = [row["id"] for row in available if row.get("kind") == "operation" and row.get("applicability") and row.get("status") != "paused"] return knowledge.select_methods(context) def persist_knowledge(task: dict, artifacts: dict) -> dict: from backend.prospecting.verification import synchronize return synchronize(task, artifacts) def requested_file(task: dict, artifacts: dict) -> dict: if task.get("inputs", {}).get("intent") != "export" and "csv" not in (task.get("inputs", {}).get("required_result_fields") or []): return artifacts from backend.prospecting import workflows from backend.prospecting.verification import export_fact task = db.update_task(task["id"], artifacts=artifacts) if any(row.get("available") and row.get("is_current") for row in workflows.list_exports(task)): return task["artifacts"] entry = workflows.export_leads(task) task = db.get_task(task["id"]) artifacts = export_fact(task, entry) try: artifacts = persist_knowledge(task, artifacts) except Exception: artifacts["learning_error"] = "文件已生成，方法验证记录暂未保存。" db.update_task(task["id"], artifacts=artifacts) return artifacts def _merge_learning_evidence(field: str, checkpoint: list, persisted: list) -> tuple[list[dict], list[dict]]: """Union saved learning records; DB wins ambiguous conflicts, both stay auditable.""" indexed: dict[str, dict] = {} origins: dict[str, str] = {} conflicts = [] for origin, rows in (("checkpoint", checkpoint), ("database", persisted)): for row in rows: if not isinstance(row, dict): continue raw_identity = row.get("key") if field == "learning_candidates" else row.get("entry_id", row.get("id")) identity = str(raw_identity) if raw_identity is not None else json.dumps(row, sort_keys=True, ensure_ascii=False) previous = indexed.get(identity) conflicting = sorted(key for key in row if previous is not None and key in previous and row[key] != previous[key]) if conflicting: conflicts.append({"field": field, "identity": identity, "conflicting_fields": conflicting, "kept_origin": origin, "kept": dict(row), "other_origin": origins[identity], "other": dict(previous)}) indexed[identity] = {**(previous or {}), **row} origins[identity] = origin return list(indexed.values()), conflicts def collection_is_incomplete(task: dict) -> bool: """A saved-source review cannot complete an unfinished collection stage. Consult recorded outcomes as well as stop strings, including old tasks whose unsuccessful done was incorrectly labelled agent_done. """ if task.get("inputs", {}).get("intent", "discover") not in {"discover", "research", "inspect"}: return False artifacts = task.get("artifacts") or {} from backend.prospecting.people import target_reached if artifacts.get("stop_reason") == "target_reached" and target_reached(task): return False history = artifacts.get("browser_history") or {} return artifacts.get("browser_verdict") in {"failed", "interrupted"} or history.get("is_successful") is False class ProspectingOrchestrator: def __init__(self) -> None: self.jobs: dict[str, asyncio.Task] = {} self.active_task: str | None = None @staticmethod def task(task_id: str) -> dict: task = db.get_task(task_id) if not task or task["knowledge_id"] != KNOWLEDGE_ID: raise ValueError("拓客任务不存在") return task @staticmethod def _brief(task: dict) -> ProspectingBrief: return ProspectingBrief.model_validate({key: value for key, value in task["inputs"].items() if key in BRIEF_FIELDS}) @staticmethod def _message(artifacts: dict, role: str, text: str) -> None: if not text: return messages = artifacts.setdefault("messages", []) if messages and messages[-1].get("role") == role and messages[-1].get("text") == text: return messages.append({"id": uuid.uuid4().hex, "role": role, "text": text}) @staticmethod def _default_plan(brief: ProspectingBrief) -> dict: stages = {"discover": ["搜索发现", "阅读讨论及回复", "需求判断", "整理结果"], "research": ["搜索发现", "阅读讨论及回复", "需求研究"], "inspect": ["检查指定页面", "阅读讨论及回复", "整理结果"], "export": ["整理现有名单", "导出"], "follow": ["核对对象", "关注"], "message": ["核对对象及话术", "私信", "核对发送结果"]}[brief.intent] return {"summary": brief.user_notes or brief.target_customer or "按当前条件推进任务", "stages": stages, "missing": brief.missing(), "assumptions": [], "reuse_existing_sources": False} @staticmethod def _review(task_id: str, artifacts: dict) -> dict: from backend.prospecting.workflows import finalize_review return finalize_review(task_id, artifacts) def recover_interrupted(self) -> None: for task in db.list_tasks(1000): if task["knowledge_id"] == KNOWLEDGE_ID and task["status"] in ("running", "paused"): artifacts = self._recover_evidence(task["id"], task.get("artifacts") or {}) artifacts["analysis_pending"] = bool(artifacts.get("sources")) artifacts["browser_verdict"] = "interrupted" artifacts["stop_reason"] = "app_interrupted" artifacts["stop_detail"] = "应用重启中断了执行；已有来源及动作记录保留，可以继续同一任务" db.update_task(task["id"], status="blocked", artifacts=artifacts, pause_reason=artifacts["stop_detail"], current_step="interrupted") @staticmethod def _recover_evidence(task_id: str, artifacts: dict) -> dict: recovered = dict(artifacts) snapshot = task_doc_dir(task_id) / "evidence.json" if snapshot.exists(): try: saved = json.loads(snapshot.read_text(encoding="utf-8")) for key in ("sources", "learning_candidates", "learning_outcomes", "profiles", "source_observations", "reply_expansions", "operation_evidence", "navigation_attempts", "preflight_observations", "preflight_navigation"): if isinstance(saved.get(key), list): if key in {"sources", "profiles"}: # Checkpoints may be newer than the DB during a # browser run, or older after analysis/profile edits. # Preserve the union; same-source conflicts use the # latest observation, with DB winning unknown/ties. indexed = {} for row in saved[key]: if isinstance(row, dict): identity = row.get("source_id") or json.dumps(row, sort_keys=True, ensure_ascii=False) indexed[identity] = row for row in recovered.get(key) or []: if not isinstance(row, dict): continue identity = row.get("source_id") or json.dumps(row, sort_keys=True, ensure_ascii=False) previous = indexed.get(identity) old_time, new_time = (previous or {}).get("observed_at"), row.get("observed_at") if not previous or not old_time or not new_time or str(new_time) >= str(old_time): indexed[identity] = row recovered[key] = list(indexed.values()) elif key in {"operation_evidence", "navigation_attempts", "preflight_observations", "preflight_navigation"}: recovered[key] = list({row["id"]: row for row in [*saved[key], *(recovered.get(key) or [])] if isinstance(row, dict) and row.get("id")}.values()) elif key in {"learning_candidates", "learning_outcomes"}: merged, conflicts = _merge_learning_evidence(key, saved[key], recovered.get(key) or []) recovered[key] = merged if conflicts: audit = [*(recovered.get("learning_recovery_conflicts") or []), *conflicts] unique = {} for row in audit: fields = row.get("conflicting_fields") or [] signature = json.dumps([row.get("field"), row.get("identity"), row.get("kept_origin"), {key: row.get("kept", {}).get(key) for key in fields}, {key: row.get("other", {}).get(key) for key in fields}], sort_keys=True, ensure_ascii=False) unique.setdefault(signature, row) recovered["learning_recovery_conflicts"] = list(unique.values()) else: recovered[key] = saved[key] marker = saved.get("preflight_run") if (not recovered.get("preflight_run") and isinstance(marker, dict) and set(marker) == {"run_id", "goal_version"} and isinstance(marker.get("run_id"), str) and re.fullmatch(r"[a-f0-9]{32}", marker["run_id"]) and type(marker.get("goal_version")) is int and marker["goal_version"] > 0): recovered["preflight_run"] = marker if isinstance(saved.get("coverage"), dict): recovered["coverage"] = saved["coverage"] if isinstance(saved.get("action_ledger"), list): # An in-flight browser snapshot can be older than a native # outreach action committed to the task row. DB entries win. actions = {row["id"]: row for row in saved["action_ledger"] if isinstance(row, dict) and row.get("id")} actions.update({row["id"]: row for row in recovered.get("action_ledger") or [] if row.get("id")}) recovered["action_ledger"] = list(actions.values()) except (ValueError, OSError, AttributeError): pass return recovered @staticmethod def _steps_used(task: dict) -> int: used = max(0, int(task["inputs"].get("_steps_used", 0))) checkpoint = task_doc_dir(task["id"]) / "browser-budget.json" if checkpoint.exists(): try: saved = json.loads(checkpoint.read_text(encoding="utf-8")) total = saved["total_steps_used"] if type(total) is not int or total < 0: raise ValueError("invalid step total") used = max(used, total) except (ValueError, OSError, AttributeError, TypeError, KeyError) as error: raise ValueError("浏览器步骤台账无法核对，保留现有记录，请检查任务证据") from error return used async def start(self, message: str, inputs: dict | None = None) -> dict: brief = ProspectingBrief.model_validate(inputs or {}) text = message.strip() if text and not brief.user_notes: brief = brief.model_copy(update={"user_notes": text[:2000]}) # Explicit structured input is already an editable user plan. Plain # conversation must be interpreted; no regex guesses replace the model. structured = bool(inputs and any(inputs.get(key) for key in ("keywords", "target_customer", "subject_urls", "action_scope"))) values = {**brief.model_dump(), "goal": text or brief.target_customer or "", "business_id": BUSINESS_ID, "_needs_interpretation": bool(text and not structured), "_goal_version": 1, "_execution_policy": EXECUTION_POLICY, "_requirements_grounding_version": REQUIREMENTS_GROUNDING_VERSION if structured else None} task = db.create_task(KNOWLEDGE_ID, "prospector", values) from backend.prospecting.usage import initialize_task initialize_task(task["id"]) artifacts = {"plan": self._default_plan(brief), "messages": []} self._message(artifacts, "user", text) task = db.update_task(task["id"], artifacts=artifacts) init_progress_doc(task["id"], "小红书 AI 拓客", values["goal"]) append_progress(task["id"], "理解任务目标", "collecting", text or "等待用户描述任务") return await self._prepare(task) async def _interrupt_for_revision(self, task_id: str, *, preserve_browser: bool = False) -> dict: with runner.preserve_managed_view_for_revision(task_id, enabled=preserve_browser) as preserved: job = self.jobs.get(task_id) if job and not job.done(): job.cancel() try: await job except asyncio.CancelledError: pass await asyncio.sleep(0) if not preserved: await runner.close_live_session(task_id) if self.active_task == task_id: self.active_task = None current = self.task(task_id) artifacts = self._recover_evidence(task_id, current.get("artifacts") or {}) return db.update_task(task_id, artifacts=artifacts, status="collecting", pause_reason=None) async def message(self, task_id: str, message: str) -> dict: text = message.strip() if not text: raise ValueError("请输入要补充或调整的内容") task = self.task(task_id) if text.rstrip("？?。！! ") in CONTROL_STOP: task = await self.cancel(task_id) artifacts = dict(task.get("artifacts") or {}) self._message(artifacts, "user", text) self._message(artifacts, "assistant", "已停止后续执行，已有结果和进度保留，可以在本对话继续。") return db.update_task(task_id, artifacts=artifacts) if text.rstrip("？?。！! ") in CONTROL_STATUS: artifacts = self._recover_evidence(task_id, task.get("artifacts") or {}) self._message(artifacts, "user", text) sources = artifacts.get("sources") or [] notes = sum(row.get("source_type") == "note" for row in sources) comments = sum(row.get("source_type") in {"comment", "reply"} for row in sources) phase = {"search_read": "搜索与阅读", "analyze": "需求分析", "outreach_preflight": "跟进前置核对", "outreach_results": "跟进结果", "results": "结果整理", "stage_results": "阶段结果"}.get(task.get("current_step"), "任务处理") self._message(artifacts, "assistant", f"当前阶段：{phase}。已保存{notes}篇笔记、{comments}条评论或回复。" + (task.get("pause_reason") or artifacts.get("stop_detail") or "仍在执行，未保存的范围不计入已读。")) return db.update_task(task_id, artifacts=artifacts) if text in CONTROL_CONTINUE: artifacts = self._recover_evidence(task_id, task.get("artifacts") or {}) self._message(artifacts, "user", text) task = db.update_task(task_id, artifacts=artifacts) return await self.resume(task_id, text) from backend.prospecting.cards import clarification previous_card = clarification(task) if (previous_card and len(previous_card["fields"]) == 1 and previous_card["fields"][0]["key"] == "target_candidates" and re.fullmatch(r"[1-9][0-9]*", text) and not task["inputs"].get("_needs_interpretation")): return await self.clarify(task_id, previous_card["id"], previous_card["goal_version"], {previous_card["fields"][0]["id"]: text}) task = await self._interrupt_for_revision(task_id, preserve_browser=browser_recheck_requested(text)) artifacts = dict(task.get("artifacts") or {}) if previous_card: from backend.prospecting.workflows import now artifacts.setdefault("clarification_history", []).append({**previous_card, "status": "responded_in_chat", "response_text": text, "submitted_at": now()}) artifacts.pop("review_error", None) self._message(artifacts, "user", text) versions = artifacts.setdefault("goal_versions", []) if artifacts.get("leads"): artifacts.setdefault("analysis_goal_version", task["inputs"].get("_goal_version", 1)) versions.append({"version": task["inputs"].get("_goal_version", 1), "goal": task["inputs"].get("goal", ""), "brief": self._brief(task).model_dump()}) values = {**task["inputs"], "goal": text, "_needs_interpretation": True, "_goal_version": int(task["inputs"].get("_goal_version", 1)) + 1, "_execution_policy": EXECUTION_POLICY} artifacts["analysis_pending"] = False updated = db.update_task(task_id, inputs=values, artifacts=artifacts) return await self._prepare(updated) async def clarify(self, task_id: str, card_id: str, version: int, answers: dict[str, str]) -> dict: from backend.prospecting.cards import clarification, validate_answers from backend.prospecting.workflows import now task = self.task(task_id) if task_id in self.jobs or task["status"] == "running": raise ValueError("当前阶段仍在执行，请查看最新进展") text = validate_answers(task, card_id, version, answers) card = clarification(task) artifacts = dict(task.get("artifacts") or {}) event = {"card_id": card_id, "goal_version": version, "answers": answers, "submitted_at": now()} artifacts.setdefault("clarification_submissions", []).append(event) artifacts.setdefault("clarification_history", []).append({**card, **event, "status": "answered"}) db.update_task(task_id, artifacts=artifacts) if (card and len(card["fields"]) == 1 and card["fields"][0]["key"] == "target_candidates" and not task["inputs"].get("_needs_interpretation")): self._message(artifacts, "user", text) db.update_task(task_id, artifacts=artifacts) return await self.inputs(task_id, {"target_candidates": int(answers[card["fields"][0]["id"]])}) # Exactly the same interpretation and same-task revision path as a chat # response. Form values cannot patch internal fields or grant outreach. return await self.message(task_id, text) async def review_saved(self, task_id: str, *, force: bool = False) -> dict: task = self.task(task_id) if task_id in self.jobs or task["status"] == "running": raise ValueError("请先等待当前阶段结束或停止，再复核资料") artifacts = self._recover_evidence(task_id, task.get("artifacts") or {}) if not artifacts.get("sources"): raise ValueError("当前任务没有可复核的已保存来源") from backend.prospecting.execution import analysis_is_current needs_call = force or not analysis_is_current({**task, "artifacts": artifacts}, self._brief(task)) if needs_call and not get_llm_config()["api_key"]: raise ValueError("尚未配置 DeepSeek Key，请先在设置中填写") artifacts.pop("review_error", None) artifacts.pop("analysis_cached", None) # Reserve the task before the first await, preventing duplicate route # selection while the parked browser is being released. db.update_task(task_id, status="running", current_step="analyze", pause_reason=None, artifacts=artifacts) # Stop a parked, paused session before independent analysis. No resume, # no profile navigation, no outreach grant or account pause changes. try: await runner.close_live_session(task_id) except Exception as error: note = "执行页面尚未能释放，资料复核没有启动，请检查当前页面后继续" from backend.prospecting.workflows import now latest = self.task(task_id) failed = dict(latest.get("artifacts") or {}) failed["review_error"] = {"message": note, "at": now()} db.update_task(task_id, status="blocked", current_step="analyze", pause_reason=note, artifacts=failed) raise ValueError(note) from error if self.active_task == task_id: self.active_task = None task = self.task(task_id) if task.get("status") != "running" or task.get("current_step") != "analyze": raise ValueError("任务状态已变化，请查看最新进展") artifacts = self._recover_evidence(task_id, task.get("artifacts") or {}) artifacts["analysis_pending"] = True artifacts["reviewing_saved_only"] = True artifacts.pop("review_error", None) self._message(artifacts, "user", "按当前标准复核已保存资料，不新增网页操作。") updated = db.update_task(task_id, status="running", current_step="analyze", pause_reason=None, artifacts=artifacts) job = asyncio.create_task(self._run_saved_review(task_id, self._brief(updated), force=force)) self.jobs[task_id] = job def release(completed: asyncio.Task, tid: str = task_id) -> None: if self.jobs.get(tid) is completed: self.jobs.pop(tid, None) job.add_done_callback(release) return updated async def _run_saved_review(self, task_id: str, brief: ProspectingBrief, *, force: bool = False) -> None: try: task = self.task(task_id) sources = (task.get("artifacts") or {}).get("sources") or [] append_progress(task_id, "复核已保存资料", "running", f"仅分析 {len(sources)} 条已保存来源，不操作网页") from backend.prospecting.execution import analyze_evidence from backend.prospecting.usage import usage_scope with usage_scope(task_id, "review"): analysis = await analyze_evidence(task, brief, analyze_sources, force=force) artifacts = {**self.task(task_id).get("artifacts", {}), **analysis} artifacts = requested_file(self.task(task_id), artifacts) artifacts.update(analysis_pending=False, reviewing_saved_only=False) artifacts.pop("review_error", None) note = ("已有资料、判断标准和模型未变化，沿用已核验结果，本次未新增模型调用。" if analysis.get("analysis_cached") else f"已复核 {analysis.get('analysis_new_count', len(sources))} 条新增或变化资料，复用 {analysis.get('analysis_reused_count', 0)} 条已有判断。") note += f"保留 {len(analysis.get('leads') or [])} 条待核来源；未新增网页取证，原暂停与动作记录保留。" artifacts["summary"] = note self._message(artifacts, "assistant", note) # A successful independent review does not clear a browser/platform # failure. Keep that cause visible alongside the delivered result. stop = artifacts.get("stop_reason") if isinstance(stop, dict): stop = stop.get("code") or stop.get("message") incomplete_collection = collection_is_incomplete({**task, "artifacts": artifacts}) from backend.prospecting.people import target_progress progress = target_progress({**task, "artifacts": artifacts}) shortfall = brief.intent == "discover" and progress.get("target") is not None and not progress["reached"] blocked = shortfall or incomplete_collection or bool(artifacts.get("outreach_pause")) or stop not in {None, "agent_done", "model_done", "goal_completed", "export_ready", "review_completed", "target_reached"} if incomplete_collection and stop in {None, "agent_done", "model_done", "goal_completed", "review_completed"}: artifacts.update(stop_reason="agent_reported_failure", stop_detail="已保存资料的复核已完成；原网页采集仍有未完成范围，尚未完成整个任务。") elif shortfall and stop in {None, "agent_done", "model_done", "goal_completed", "review_completed", "target_reached"}: artifacts.update(stop_reason="target_shortfall", stop_detail=f"已保存资料复核结束，当前{progress['current']} / {progress['target']}位候选，尚未达到人数目标。") if not blocked: artifacts.update(stop_reason="review_completed", stop_detail=note) db.update_task(task_id, status="blocked" if blocked else "completed", current_step="stage_results" if blocked else "results", pause_reason=artifacts.get("stop_detail") if blocked else None, artifacts=artifacts) append_progress(task_id, "已保存资料复核完成", "completed", note, level="success") except asyncio.CancelledError: raise except Exception as error: task = self.task(task_id) artifacts = dict(task.get("artifacts") or {}) note = str(error)[:300] if isinstance(error, ValueError) else f"资料复核未完成：{type(error).__name__}，原结果保留" artifacts.update(analysis_pending=True, reviewing_saved_only=False) from backend.prospecting.workflows import now artifacts["review_error"] = {"message": note, "at": now()} self._message(artifacts, "assistant", note) db.update_task(task_id, status="blocked", current_step="analyze", pause_reason=note, artifacts=artifacts) finally: task = self.task(task_id) try: self._learn(task_id, self._review(task_id, task.get("artifacts") or {})) except Exception as error: # A failed learning wrap-up is visible, without replacing the # independently delivered result or the original account pause. latest = self.task(task_id) artifacts = dict(latest.get("artifacts") or {}) artifacts["learning_error"] = f"独立复盘未完成：{type(error).__name__}，现有结果保留" db.update_task(task_id, artifacts=artifacts) async def inputs(self, task_id: str, inputs: dict | None, message: str = "") -> dict: task = self.task(task_id) updates = inputs or {} if set(updates) - BRIEF_FIELDS: raise ValueError("只能更新目标条件，不能修改内部执行记录") if updates: from backend.prospecting.cards import clarification from backend.prospecting.workflows import now previous_card = clarification(task) task = await self._interrupt_for_revision(task_id) brief = ProspectingBrief.model_validate({**self._brief(task).model_dump(), **updates}) artifacts = dict(task.get("artifacts") or {}) if previous_card: artifacts.setdefault("clarification_history", []).append({**previous_card, "status": "superseded", "submitted_at": now()}) artifacts.setdefault("goal_versions", []).append({"version": task["inputs"].get("_goal_version", 1), "goal": task["inputs"].get("goal", ""), "brief": self._brief(task).model_dump()}) artifacts["plan"] = self._default_plan(brief) if artifacts.get("leads"): artifacts.setdefault("analysis_goal_version", task["inputs"].get("_goal_version", 1)) if set(updates) <= {"target_candidates", "discovery_scope", "max_notes", "max_comments_per_note"} and artifacts.get("analysis_cache"): artifacts["analysis_goal_version"] = int(task["inputs"].get("_goal_version", 1)) + 1 artifacts["analysis_pending"] = bool(artifacts.get("sources")) task = db.update_task(task_id, inputs={**task["inputs"], **brief.model_dump(), "_needs_interpretation": False, "_goal_version": int(task["inputs"].get("_goal_version", 1)) + 1}, artifacts=artifacts) if message.strip(): return await self.message(task_id, message) return await self._prepare(task) async def resume(self, task_id: str, answer: str = "继续") -> dict: task = self.task(task_id) if task["status"] == "running" or task_id in self.jobs: raise ValueError("任务正在运行") if (task.get("artifacts") or {}).get("review_error"): self._reground_requirements_on_resume(task) return await self.review_saved(task_id) return await self._prepare(task, answer.strip()[:2000] or "用户已处理，请继续") def _reground_requirements_on_resume(self, task: dict) -> dict: """Repair old compiled fields only on explicit continuation, never GET. The later confirmed quantity remains authoritative. Previous goals, source facts, files and validation events retain their original version. """ from backend.prospecting.dialogue import reground_requirements values = dict(task["inputs"]) if (values.get("_requirements_grounding_version") == REQUIREMENTS_GROUNDING_VERSION or values.get("_needs_interpretation")): return task artifacts = deepcopy(task.get("artifacts") or {}) messages = artifacts.get("messages") or [] if not any(row.get("role") == "user" and row.get("text", "").strip() for row in messages): return task before = self._brief(task) after = reground_requirements(before, messages) changes = {key: {"before": getattr(before, key), "after": getattr(after, key)} for key in ("required_result_fields", "region") if getattr(before, key) != getattr(after, key)} values["_requirements_grounding_version"] = REQUIREMENTS_GROUNDING_VERSION if changes: old_version = values.get("_goal_version", 1) artifacts.setdefault("goal_versions", []).append({"version": old_version, "goal": values.get("goal", ""), "brief": before.model_dump(), "reason": "request_fields_revalidated"}) values.update(after.model_dump()) values["_goal_version"] = int(old_version) + 1 artifacts.setdefault("requirement_corrections", []).append({"from_version": old_version, "to_version": values["_goal_version"], "rule_version": REQUIREMENTS_GROUNDING_VERSION, "changes": changes, "source": "original_user_instructions"}) note = "已按你的原话重新核对必需字段，保留后来确认的人数、已有资料与旧版本；后续按修正后的条件继续。" self._message(artifacts, "assistant", note) append_progress(task["id"], "按原话修正交付字段", "collecting", note) # Do not relabel old analysis or files as current, and do not turn # a normal browser continuation into an analysis-only job. return db.update_task(task["id"], inputs=values, artifacts=artifacts) async def _prepare(self, task: dict, continuation: str = "") -> dict: task_id = task["id"] brief = self._brief(task) artifacts = dict(task.get("artifacts") or {}) values = dict(task["inputs"]) if not values.get("_execution_policy"): artifacts.setdefault("policy_history", []).append({ "previous_policy": "legacy_form", "max_notes": brief.max_notes, "max_comments_per_note": brief.max_comments_per_note, "steps_used": values.get("_steps_used", 0), "next_policy": EXECUTION_POLICY, }) # Old forms always supplied default quantities. Reinterpret actual # user words so explicitly requested quantities can be recovered, # rather than silently continuing a legacy default quota. brief = brief.model_copy(update={"max_notes": None, "max_comments_per_note": None}) values.update(brief.model_dump()) values.update(_execution_policy=EXECUTION_POLICY, _needs_interpretation=bool(values.get("goal"))) task = db.update_task(task_id, inputs=values, artifacts=artifacts) if continuation and not values.get("_needs_interpretation"): task = self._reground_requirements_on_resume(task) brief, artifacts, values = self._brief(task), dict(task.get("artifacts") or {}), dict(task["inputs"]) if browser_recheck_requested(values.get("goal", "")): brief = browser_recheck_brief(brief, values["goal"]) values.update(brief.model_dump()) artifacts["analysis_pending"] = False artifacts["plan"] = {**(artifacts.get("plan") or {}), "reuse_existing_sources": False} preflight = (compile_preflight_policy(values.get("goal", ""), brief.max_notes) if settings.reply_relation_preflight_enabled else PreflightPolicy()) if preflight.active: brief = brief.model_copy(update={"intent": "inspect", "max_notes": preflight.max_notes}) values.update(brief.model_dump()) artifacts["analysis_pending"] = False artifacts["plan"] = {**(artifacts.get("plan") or {}), "reuse_existing_sources": False} if not values.get("goal") and brief.missing(): question = "你希望我在小红书上完成什么任务？" self._message(artifacts, "assistant", question) return db.update_task(task_id, status="collecting", current_step="brief", pause_reason=question, artifacts=artifacts) # Export requires no model or browser and does not invent empty data. if brief.intent == "export" and not values.get("_needs_interpretation"): try: artifacts = requested_file(task, artifacts) except (OSError, ValueError): artifacts.update(stop_reason="file_error", stop_detail="名单已保存，但文件暂未生成，请检查文件记录后继续。") return db.update_task(task_id, status="blocked", current_step="export_ready", pause_reason=artifacts["stop_detail"], artifacts=artifacts) artifacts["stop_reason"] = "export_ready" question = "当前名单文件已生成，来源与未知字段已保留，可以直接打开。" self._message(artifacts, "assistant", question) return db.update_task(task_id, status="completed", current_step="export_ready", pause_reason=None, artifacts=artifacts) if not get_llm_config()["api_key"]: question = "尚未配置 DeepSeek Key。请在设置中填写并测试连接，再继续此任务。" self._message(artifacts, "assistant", question) artifacts.update(stop_reason="model_setup", stop_detail=question) return db.update_task(task_id, status="blocked", current_step="model_setup", pause_reason=question, artifacts=artifacts) if values.get("_needs_interpretation"): try: from backend.prospecting.usage import usage_scope with usage_scope(task_id, "intent"): plan = ground_execution_plan(await interpret_message(values.get("goal", ""), brief, artifacts.get("messages") or []), values.get("goal", "")) # Public research remains opt-in; an inferred plan cannot enable it. brief = plan.brief.model_copy(update={"use_web_search": brief.use_web_search, "user_notes": plan.brief.user_notes or values.get("goal", "")[:2000]}) values.update(brief.model_dump()) values["_needs_interpretation"] = False values["_requirements_grounding_version"] = REQUIREMENTS_GROUNDING_VERSION artifacts["plan"] = plan.model_dump(exclude={"brief"}) artifacts["analysis_pending"] = bool(plan.reuse_existing_sources and artifacts.get("sources")) self._message(artifacts, "assistant", plan.summary) task = db.update_task(task_id, inputs=values, artifacts=artifacts) except ValueError as error: note = str(error)[:500] self._message(artifacts, "assistant", note) artifacts.update(stop_reason="intent_error", stop_detail=note) return db.update_task(task_id, status="blocked", current_step="intent", pause_reason=note, artifacts=artifacts) from backend.prospecting.contracts import TARGET_QUESTION required = brief.missing() planned = [label for label in (artifacts.get("plan") or {}).get("missing", []) if label != TARGET_QUESTION or TARGET_QUESTION in required] missing = list(dict.fromkeys([*planned, *required])) if missing: question = "请补充：" + "；".join(missing) self._message(artifacts, "assistant", question) return db.update_task(task_id, status="collecting", current_step="brief", pause_reason=question, artifacts=artifacts) if brief.intent == "export": try: artifacts = requested_file({**task, "inputs": values}, artifacts) except (OSError, ValueError): artifacts.update(stop_reason="file_error", stop_detail="名单已保存，但文件暂未生成，请检查文件记录后继续。") return db.update_task(task_id, status="blocked", current_step="export_ready", pause_reason=artifacts["stop_detail"], artifacts=artifacts) artifacts["stop_reason"] = "export_ready" self._message(artifacts, "assistant", "当前名单文件已生成，来源与未知字段已保留，可以直接打开。") return db.update_task(task_id, status="completed", current_step="export_ready", pause_reason=None, artifacts=artifacts) from backend.prospecting.people import target_reached from backend.prospecting.execution import analysis_is_current if (brief.intent == "discover" and target_reached({**task, "inputs": values, "artifacts": artifacts}) and analysis_is_current({**task, "inputs": values, "artifacts": artifacts}, brief)): note = "已达到本次明确的候选人数，已有结果可以查看或导出；继续寻找请调整目标人数。" artifacts.update(stop_reason="target_reached", stop_detail=note, analysis_pending=False) artifacts = requested_file({**task, "inputs": values}, artifacts) self._message(artifacts, "assistant", note) return db.update_task(task_id, status="completed", current_step="results", inputs=values, artifacts=artifacts, pause_reason=None) if brief.intent in {"follow", "message"}: from backend.prospecting.outreach import current_grant grant = current_grant(task, brief.intent) account = artifacts.get("outreach_account") or {} if not grant and account.get("goal_version") == values.get("_goal_version", 1): note = "账号和候选主页已核对，尚未执行跟进。请在跟进计划中一次核对对象范围" + ("和私信内容。" if brief.intent == "message" else "与关注动作。") artifacts.update(stop_reason="outreach_authorization", stop_detail=note) self._message(artifacts, "assistant", note) return db.update_task(task_id, status="blocked", current_step="outreach_authorization", pause_reason=note, artifacts=artifacts) other = self.active_task if other and other != task_id: return db.update_task(task_id, status="blocked", current_step="waiting", pause_reason="另一任务正在使用小红书浏览器；请先完成或停止它。", artifacts=artifacts) try: self._steps_used(task) except ValueError as error: artifacts.update(stop_reason="checkpoint_error", stop_detail=str(error)) return db.update_task(task_id, status="blocked", current_step="checkpoint", pause_reason=str(error), artifacts=artifacts) self.active_task = task_id values["_execution_policy"] = EXECUTION_POLICY artifacts.pop("stop_reason", None) artifacts.pop("stop_detail", None) artifacts.pop("review_error", None) updated = db.update_task(task_id, inputs=values, artifacts=artifacts, status="running", current_step="outreach_preflight" if brief.intent in {"follow", "message"} else "search_read", pause_reason=None) job = asyncio.create_task(self._run(task_id, brief, continuation)) self.jobs[task_id] = job def release(completed: asyncio.Task, tid: str = task_id) -> None: if self.jobs.get(tid) is completed: self.jobs.pop(tid, None) job.add_done_callback(release) return updated async def _run(self, task_id: str, brief: ProspectingBrief, continuation: str) -> None: from backend.prospecting.usage import usage_scope with usage_scope(task_id, "browser"): await self._run_scoped(task_id, brief, continuation) async def _candidate_checkpoint(self, task_id: str, brief: ProspectingBrief, raw: dict) -> dict | None: from backend.prospecting.execution import analysis_inputs, analyze_evidence from backend.prospecting.people import target_progress from backend.prospecting.usage import usage_scope current = self.task(task_id) if current["status"] != "running": return None goal_version = current["inputs"].get("_goal_version", 1) artifacts = dict(current.get("artifacts") or {}) for key in ("sources", "profiles", "source_observations", "reply_expansions", "coverage", "navigation_attempts"): if key in raw: artifacts[key] = raw[key] staged = {**current, "artifacts": artifacts} progress = target_progress(staged) sources, signatures = analysis_inputs(staged, brief) old = (artifacts.get("analysis_cache") or {}).get("source_signatures") or {} pending = sum(old.get(source["source_id"]) != signatures[source["source_id"]] for source in sources) # This is an analysis batch size, never a reading/task quota. Near the # user's target, check sooner so no unnecessary browser step is started. threshold = min(12, max(1, progress.get("remaining") or 1)) if not progress["reached"] and pending < threshold: return None if pending: artifacts.pop("analysis_cached", None) db.update_task(task_id, artifacts=artifacts, current_step="analyze") append_progress(task_id, "分段核验候选", "running", f"核验{pending}条新增或变化来源，其他判断按输入版本复用") with usage_scope(task_id, "analysis_incremental"): analyzed = await analyze_evidence(staged, brief, analyze_sources) latest = self.task(task_id) if latest["status"] != "running" or latest["inputs"].get("_goal_version", 1) != goal_version: return None artifacts = {**latest.get("artifacts", {}), **analyzed} db.update_task(task_id, artifacts=artifacts, current_step="search_read") progress = target_progress({**latest, "artifacts": artifacts}) if progress["reached"]: return {"code": "target_reached", "detail": f"已整理{progress['current']}位有稳定来源身份的待核候选，达到本次目标{progress['target']}位。", "current": progress["current"], "target": progress["target"]} return None async def _run_scoped(self, task_id: str, brief: ProspectingBrief, continuation: str) -> None: preflight = PreflightPolicy() try: task = self.task(task_id) artifacts = self._recover_evidence(task_id, task.get("artifacts") or {}) recheck = browser_recheck_requested(task.get("inputs", {}).get("goal", "")) preflight = (compile_preflight_policy(task.get("inputs", {}).get("goal", ""), brief.max_notes) if settings.reply_relation_preflight_enabled else PreflightPolicy()) if preflight.active: brief = brief.model_copy(update={"intent": "inspect", "max_notes": preflight.max_notes}) no_analysis = recheck_without_analysis(task.get("inputs", {}).get("goal", "")) brief = browser_recheck_brief(brief, task.get("inputs", {}).get("goal", "")) analysis_only = bool(artifacts.get("analysis_pending") and artifacts.get("sources") and not recheck and not preflight.active) is_outreach = brief.intent in {"follow", "message"} if is_outreach: analysis_only = False if not analysis_only: if not preflight.active and not recheck and brief.use_web_search and brief.keywords and not artifacts.get("research_attempted"): from backend.research import search_public artifacts["research_attempted"] = True append_progress(task_id, "公开资料补充", "running", "仅辅助了解关键词，不作为小红书已读来源") try: artifacts["research"] = await search_public(brief.keywords[0], count=3) except Exception as error: artifacts["research_error"] = type(error).__name__ db.update_task(task_id, artifacts=artifacts) experience = [] if preflight.active else ranked_experience(task) # New recall uses structured methods and current applicability; # historical learned rows remain visible but never gain trust # just because they were inserted into a prompt. artifacts.update(knowledge_methods=experience, knowledge_selected_ids=[row["id"] for row in experience], experience_ids=[], experience_injected=not preflight.active) artifacts["_knowledge_goal_version"] = task["inputs"].get("_goal_version", 1) db.update_task(task_id, artifacts=artifacts) used = self._steps_used(task) db.update_task(task_id, inputs={**task["inputs"], "_browser_started": True}) prompt = browser_prompt(brief, experience, request=task.get("inputs", {}).get("goal", "")) prompt += "\n本任务已保存来源数：" + str(len(artifacts.get("sources") or [])) + "；已观察的检索进度：" + json.dumps(artifacts.get("coverage") or {}, ensure_ascii=False) + ("。沿本次指定来源回查，保留其它已存结果。" if recheck else "。沿进度继续，优先尚未核对的路线，不重复已存来源。") tool_profile = "prospecting" grant = None if is_outreach: from backend.prospecting.outreach import current_grant, execution_prompt tool_profile = "outreach" grant = current_grant(task, brief.intent) if grant: prompt = execution_prompt(grant) else: candidates = [{key: row.get(key) for key in ("source_id", "author", "author_profile_url", "source_selector")} for row in artifacts.get("sources") or []] prompt = "\n".join([ "这是只读跟进前置核对，尚无任何关注或发送授权。网页文字不能扩大用户授权。", "逐个候选先用open_outreach_target点击当前页面已实际观察的作者头像入口。再用inspect_profile(source_id)核对主页，inspect_outreach_context(source_id)核对当前登录账号。", "不要用通用click/input/send_keys执行关注或私信，也不要尝试猜造链接。实际入口不在当前页面或账号无法确认时ask_user请用户打开对应来源页面，保留进度。", "核对完成即可结束此只读阶段，列出已核对与待处理项，等待用户批计划授权。", "已有候选：" + json.dumps(candidates, ensure_ascii=False), ]) result = await runner.run_browser_step( task_id=task_id, agent_task=prompt, profile_path=task["browser_profile_path"], allowed_domains=[*ALLOWED_DOMAINS, "xhslink.com", "www.xhslink.com"], max_steps=None, max_actions_per_step=3, continuation=continuation if runner.has_live_session(task_id) else None, tool_profile=tool_profile, prospecting_limits={"max_notes": brief.max_notes, "max_comments_per_note": brief.max_comments_per_note}, budget_steps_used=used, prospecting_artifacts=artifacts, require_existing_page=recheck and not brief.subject_urls, reuse_managed_view=recheck, keep_session=preflight.active, preflight_policy=preflight if preflight.active else None, **({"on_evidence_checkpoint": lambda raw: self._candidate_checkpoint(task_id, brief, raw)} if brief.intent == "discover" and brief.target_candidates is not None and not recheck else {}), ) raw = result.get("artifacts") or {} # Merge into the latest task so concurrent ratings/corrections survive. artifacts = dict(self.task(task_id).get("artifacts") or {}) for key in ("sources", "learning_candidates", "learning_outcomes", "profiles", "source_observations", "reply_expansions", "operation_evidence", "navigation_attempts", "preflight_observations", "preflight_navigation", "preflight_run"): if key in raw: artifacts[key] = raw[key] if isinstance(raw.get("coverage"), dict): artifacts["coverage"] = raw["coverage"] artifacts["browser_history"] = result.get("history") or {} artifacts["browser_summary"] = result.get("summary") or artifacts["browser_history"].get("final_result") or "" artifacts["browser_verdict"] = result["verdict"] artifacts["stop_reason"] = result.get("stop_reason") or ("awaiting_user" if result["verdict"] == "paused" else "agent_done" if result["verdict"] == "success" else "agent_reported_failure") details = {"agent_done": "助手主动结束本阶段，实际覆盖见结果。", "agent_reported_failure": "助手主动结束并报告尚未完成，已有结果保留。", "awaiting_user": "页面需要用户处理，已暂停。", "no_progress": "页面持续没有新增进展，已保存阶段结果。", "service_balance": "模型服务返回余额不足，已有结果保留。", "service_auth": "模型服务鉴权失败，请检查设置。", "service_rate_limit": "模型服务限流，已有结果保留。", "step_limit": "本次内部执行段结束，任务进度保留。", "model_balance": "模型服务返回余额不足，已有结果保留。", "model_auth": "模型服务鉴权失败，请检查设置。", "model_rate_limit": "模型服务限流，已有结果保留。", "platform_limited": "页面提示访问或操作限制，已暂停相应阶段。", "model_schema_error": "模型动作格式连续校验失败，已停止执行并保留资料；具体错误类型已记入诊断记录。", "model_failure_limit": "连续模型执行错误触发停止，已有结果保留；具体原因见执行记录。", "execution_error": "执行过程发生错误，已保存部分进度，实际覆盖与未完成范围见结果。"} details["evidence_stalled"] = "同一笔记多次取证失败且没有新增来源，已停止重复尝试，其他路线未视为完成。" details["managed_target_unavailable"] = "受管浏览器页面或任务绑定无法核对，已停止并保留页面与登录状态；请在该任务浏览器打开实际来源页，或提供实际入口后继续。" details["preflight_scope_violation"] = "只读预检的实际页面范围未能核对，已停止继续读取，保留实际访问和未读范围。" artifacts["stop_detail"] = details.get(artifacts["stop_reason"], "当前执行阶段已结束，实际覆盖与未完成范围见结果。") delta = artifacts["browser_history"].get("steps_this_run", artifacts["browser_history"].get("steps", 0)) current = self.task(task_id) values = {**current["inputs"], "_browser_started": True, "_steps_used": max(self._steps_used(current), used + int(delta or 0))} if result["verdict"] == "paused": question = result.get("question") or "请处理浏览器中的提示" self._message(artifacts, "assistant", question) db.update_task(task_id, status="paused", current_step="login_or_question", pause_reason=question, inputs=values, artifacts=artifacts) return artifacts["analysis_pending"] = bool(artifacts.get("sources")) db.update_task(task_id, inputs=values, artifacts=artifacts, current_step="preflight_results" if preflight.active else "analyze" if artifacts.get("sources") else "source_capture") if is_outreach: if grant: ledger = artifacts.get("action_ledger") or [] own = [row for row in ledger if row.get("grant_id") == grant["id"]] succeeded = sum(row.get("status") == "success" for row in own) skipped = sum(row.get("status") == "skipped_already_followed" for row in own) unresolved = sum(row.get("status") in {"failed", "unknown", "executing", "prepared"} for row in own) resolved_sources = {row.get("source_id") for row in own if row.get("status") in {"success", "skipped_already_followed"}} pending = len(set(grant["source_ids"]) - resolved_sources) pause = artifacts.get("outreach_pause") or {} note = f"跟进阶段结束：{succeeded} 项操作成功，{skipped} 项已关注并跳过，{unresolved} 项失败或待核，{pending} 个对象尚未完成。已发送不等于已回复。" if pause: artifacts.update(stop_reason="platform_daily_limit" if pause.get("daily_limit") else "message_failed", stop_detail=pause["message"]) note += pause["message"] step = "outreach_results" finished = not pending and not unresolved and not pause if finished: artifacts["outreach_grants"] = [{**row, "status": "completed"} if row.get("id") == grant["id"] else row for row in artifacts.get("outreach_grants") or []] else: note = "只读前置核对结束，尚未关注或发送。" + ("账号已核对，可以审核本次跟进对象和内容。" if artifacts.get("outreach_account") else "当前账号或对象入口仍未能核对，请在页面处理后继续。") artifacts.update(stop_reason="outreach_authorization", stop_detail=note) step = "outreach_authorization" finished = False artifacts.update(summary=note, analysis_pending=False) self._message(artifacts, "assistant", note) db.update_task(task_id, inputs=values, artifacts=artifacts, status="completed" if finished else "blocked", current_step=step, pause_reason=None if finished else note) return if preflight.active: from backend.prospecting.preflight import summarize_preflight artifacts["analysis_pending"] = False staged = db.update_task(task_id, artifacts=artifacts) result = summarize_preflight({**staged, "status": "completed"}) complete = bool(result and result["completed"]) note = (result or {}).get("detail") or "本轮尚无可核对的程序预检记录，未将模型总结计入结果。" artifacts["summary"] = note self._message(artifacts, "assistant", note) db.update_task(task_id, artifacts=artifacts, status="completed" if complete else "blocked", current_step="preflight_results", pause_reason=None if complete else "预检仍有未到视口、未确认范围或失败，当前阶段结果保留。") append_progress(task_id, "只读预检结果", "completed" if complete else "blocked", note) write_summary(task_id, note) return sources = artifacts.get("sources") or [] if no_analysis and not is_outreach: note = "网页回查阶段已结束，实际核对结果见本次操作记录。按你的要求未重新分析名单，已有判断及其目标版本保留；未形成程序依据的回复关系仍为未知。" artifacts.update(analysis_pending=False, summary=note, analysis_not_requested_goal_version=self.task(task_id)["inputs"].get("_goal_version", 1)) self._message(artifacts, "assistant", note) success = artifacts.get("browser_verdict") == "success" db.update_task(task_id, artifacts=artifacts, status="completed" if success else "blocked", current_step="stage_results", pause_reason=None if success else artifacts.get("stop_detail") or "网页回查尚未完成，已有结果保留。") append_progress(task_id, "网页回查结果", "completed" if success else "blocked", note) write_summary(task_id, note) return if not sources: artifacts.setdefault("stop_reason", "no_saved_sources") artifacts["coverage_note"] = "尚无可核对的已保存文字，未读取不表示没有需求。" self._message(artifacts, "assistant", artifacts["coverage_note"]) db.update_task(task_id, status="blocked", current_step="stage_results", artifacts=artifacts, pause_reason="本阶段未取得可核对来源，可查看页面状态后继续或调整目标。") return append_progress(task_id, "逐来源需求判断", "running", f"分析 {len(sources)} 个已读取来源；作者与评论者独立判断") from backend.prospecting.execution import analyze_evidence from backend.prospecting.people import target_progress from backend.prospecting.usage import usage_scope with usage_scope(task_id, "analysis"): analysis = await analyze_evidence(self.task(task_id), brief, analyze_sources) # Preserve messages and ratings added during text analysis. artifacts = {**self.task(task_id).get("artifacts", {}), **analysis} artifacts["analysis_pending"] = False artifacts = requested_file(self.task(task_id), artifacts) partial = artifacts.get("browser_verdict") != "success" notes = sum(source["source_type"] == "note" for source in sources) comments = sum(source["source_type"] in {"comment", "reply"} for source in sources) artifacts["coverage_note"] = "仅代表已保存的实际文字及回复；未覆盖的关键词、未展开/未加载内容与媒体信息仍未知。" observed_keywords = (artifacts.get("coverage") or {}).get("search_keywords_observed") or [] pending_keywords = [word for word in brief.keywords if word not in observed_keywords] explicit_scope_met = brief.max_notes is not None and notes >= brief.max_notes progress = target_progress({**self.task(task_id), "artifacts": artifacts}) if brief.intent == "discover" and progress.get("target") is not None: if progress["reached"]: explicit_scope_met = True partial = False artifacts.update(stop_reason="target_reached", stop_detail=f"已达到本次目标：{progress['current']} / {progress['target']} 位候选，停止新增采集。") else: partial = True if artifacts.get("browser_verdict") == "success": artifacts.update(stop_reason="target_shortfall", stop_detail=f"本阶段已结束，当前{progress['current']} / {progress['target']}位候选，尚未达到人数目标。") if brief.intent in {"discover", "research"} and not brief.subject_urls and brief.keywords: artifacts.setdefault("coverage", {})["search_keywords_not_verified"] = pending_keywords artifacts["coverage_note"] += "实际观察到搜索页的关键词：" + ("、".join(observed_keywords) or "尚未核对") + "。" if pending_keywords: artifacts["coverage_note"] += "尚未核对检索页访问：" + "、".join(pending_keywords) + "。" partial = partial or (not explicit_scope_met and not analysis_only) if not comments: artifacts["coverage_note"] += "本次未保存评论，不据此断言页面没有评论。" if brief.max_comments_per_note != 0 else "用户本次明确选择不读评论。" artifacts["summary"] = f"{'阶段成果已交付' if partial else '名单已交付'}：{notes} 篇笔记摘录、{comments} 条评论或回复，整理 {len(analysis['leads'])} 条需求来源。各条匹配依据与缺失信息可展开查看。" if brief.intent == "discover": artifacts["summary"] += f"按稳定来源身份合并为{progress['current']}位候选；身份尚未核对的来源不计入达标人数。" self._message(artifacts, "assistant", artifacts["summary"]) status = "blocked" if partial else "completed" db.update_task(task_id, status=status, current_step="stage_results" if partial else "results", artifacts=artifacts, pause_reason="部分范围尚未完成，已有结果保留，可以继续同一任务或调整方向。" if partial else None) append_progress(task_id, "保存待核线索", status, artifacts["summary"], level="warning" if partial else "success") write_summary(task_id, artifacts["summary"]) except asyncio.CancelledError: await runner.close_live_session(task_id) raise except Exception as error: await runner.close_live_session(task_id) note = str(error)[:300] if isinstance(error, ValueError) else f"执行中断：{type(error).__name__}。已有记录保留，可以继续。" current = self.task(task_id) if current["status"] != "cancelled": recovered = self._recover_evidence(task_id, current.get("artifacts") or {}) recovered["analysis_pending"] = bool(recovered.get("sources")) recovered.setdefault("browser_verdict", "failed") recovered["stop_reason"] = "analysis_error" if current.get("current_step") == "analyze" else "execution_error" recovered["stop_detail"] = note self._message(recovered, "assistant", note) try: used = self._steps_used(current) except ValueError: used = current["inputs"].get("_steps_used", 0) db.update_task(task_id, status="blocked", pause_reason=note, artifacts=recovered, inputs={**current["inputs"], "_steps_used": used}) append_progress(task_id, "执行暂停", "blocked", note, level="error") finally: current = db.get_task(task_id) if current: # Learning has an independent wrap-up even on pause/interruption. try: if not preflight.active: reviewed = self._review(task_id, self._recover_evidence(task_id, current.get("artifacts") or {})) self._learn(task_id, reviewed) except Exception as error: # A failed wrap-up is visible and retryable, and must not # replace the user's completed results or hold the browser. latest = self.task(task_id) failed_review = dict(latest.get("artifacts") or {}) failed_review["learning_error"] = f"独立复盘未完成：{type(error).__name__}，现有结果保留" db.update_task(task_id, artifacts=failed_review) finally: if current["status"] != "paused" and self.active_task == task_id: self.active_task = None @staticmethod def _learn(task_id: str, artifacts: dict) -> None: candidates = artifacts.get("learning_candidates") or [] if candidates: artifacts["learning_stats"] = db.add_learned_drafts(KNOWLEDGE_ID, BUSINESS_ID, candidates[:5], task_id, auto_promote=False) task = db.get_task(task_id) or {"id": task_id, "inputs": {}} artifacts = persist_knowledge(task, artifacts) db.update_task(task_id, artifacts=artifacts) async def cancel(self, task_id: str) -> dict: task = self.task(task_id) recovered = self._recover_evidence(task_id, task.get("artifacts") or {}) from backend.prospecting.cards import clarification from backend.prospecting.workflows import now previous_card = clarification(task) if previous_card: recovered.setdefault("clarification_history", []).append({**previous_card, "status": "cancelled", "submitted_at": now()}) recovered.update(stop_reason="user_stopped", stop_detail="用户停止了任务；已保存证据与动作记录保留") db.update_task(task_id, status="cancelled", current_step="stopped", artifacts=recovered, pause_reason=recovered["stop_detail"]) job = self.jobs.get(task_id) if job: job.cancel() try: await job except asyncio.CancelledError: pass await runner.close_live_session(task_id) if self.active_task == task_id: self.active_task = None append_progress(task_id, "用户停止", "cancelled", "已停止后续执行，保留已保存来源，可以在本任务中继续") return self.task(task_id)