import asyncio import csv import inspect import io import json import math import re import urllib.parse import uuid from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Literal from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator from gfa_core.mas.agents.base import AgentMessage, AgentState from gfa_core.mas.agents.intent_v2 import IntentAgentV2 from gfa_core.mas.agents.interpretation_evidence import build_evidence_report from gfa_core.mas.context_index import build_project_context from gfa_core.mas.dual_kb_qa import query_dual_kb_qa from gfa_core.mas.llm import LLMManager from gfa_core.mas.mutation_surface import ToolSurfaceEntry from gfa_core.mas.stage_capabilities import get_stage_capability from gfa_core.mas.tools.stage_status import SUCCESS_STATUSES from gfa_core.mas.tools.stage_tools import stage_registry from gfa_core.mas.workflow_catalog import load_rnaseq_business_context from gfa_core.mas.workflow_models import GoalLedger @dataclass(frozen=True) class _ConversationTool: description: str args_model: type[BaseModel] handler: Callable availability: Callable[[], bool] | None permission: Literal["read_only", "pure"] max_result_chars: int class ConversationToolRegistry: def __init__(self, progress_callback: Callable[[dict[str, Any]], None] | None = None) -> None: self._tools: dict[str, _ConversationTool] = {} self._permission_guard: Callable[[str], None] | None = None self._progress_callback = progress_callback self._analysis_handoff_resolver: Callable[..., str] | None = None def set_analysis_handoff_resolver(self, resolver: Callable[..., str]) -> None: self._analysis_handoff_resolver = resolver def resolve_analysis_handoff_request( self, query: str, *, workflow_id: str | None = None, guidance: bool = False, language: str = "zh", ) -> str: text = str(query or "").strip() if self._analysis_handoff_resolver is None: return text resolved = str( self._analysis_handoff_resolver( text, workflow_id=workflow_id, guidance=guidance, language=language, ) or "" ).strip() return resolved or text def _notify_progress(self, event: dict[str, Any]) -> None: if self._progress_callback is None: return try: self._progress_callback(event) except Exception: pass def set_permission_guard(self, guard: Callable[[str], None]) -> None: self._permission_guard = guard def register( self, *, name: str, description: str, args_model: type[BaseModel], handler: Callable, availability: Callable[[], bool] | None = None, permission: Literal["read_only", "pure"], max_result_chars: int = 24_000, ) -> None: if permission not in {"read_only", "pure"}: raise ValueError("permission must be 'read_only' or 'pure'") if max_result_chars <= 0: raise ValueError("max_result_chars must be positive") if name in self._tools: raise ValueError(f"tool already registered: {name}") self._tools[name] = _ConversationTool( description=description, args_model=args_model, handler=handler, availability=availability, permission=permission, max_result_chars=max_result_chars, ) def definitions(self) -> list[dict[str, Any]]: return [ { "type": "function", "function": { "name": name, "description": tool.description, "parameters": tool.args_model.model_json_schema(), }, } for name, tool in self._tools.items() ] def policy_inventory(self) -> tuple[ToolSurfaceEntry, ...]: return tuple( ToolSurfaceEntry( name=name, permission=tool.permission, handler=f"{tool.handler.__module__}.{tool.handler.__qualname__}", ) for name, tool in sorted(self._tools.items()) ) async def dispatch(self, name: str, raw_arguments: str) -> str: tool = self._tools.get(name) if tool is None: return self._error("unknown_tool", f"unknown tool: {name}") if self._permission_guard is not None: try: self._permission_guard(tool.permission) except Exception as exc: return self._error("evidence_root_read_only", str(exc)) if tool.availability is not None: try: available = tool.availability() except Exception: available = False if not available: return self._error("tool_unavailable", f"tool unavailable: {name}") try: arguments = tool.args_model.model_validate_json(raw_arguments or "{}") except ValidationError as exc: return self._error("invalid_arguments", str(exc)) self._notify_progress({"stage": "tool_started", "tool": name}) try: result = tool.handler(arguments) if inspect.isawaitable(result): result = await result return self._success(result, tool.max_result_chars) except Exception as exc: return self._error("tool_error", str(exc)) finally: self._notify_progress({"stage": "tool_completed", "tool": name}) @staticmethod def _error(code: str, message: str) -> str: return json.dumps( {"ok": False, "error": {"code": code, "message": message}}, ensure_ascii=False, ) @staticmethod def _success(data: Any, max_result_chars: int) -> str: data_json = json.dumps(data, ensure_ascii=False) if len(data_json) > max_result_chars: return json.dumps( { "ok": True, "data": data_json[:max_result_chars], "truncated": True, }, ensure_ascii=False, ) return json.dumps({"ok": True, "data": data}, ensure_ascii=False) def _validate_job_id(value: str) -> str: if value and (Path(value).is_absolute() or "/" in value or "\\" in value or value in {".", ".."}): raise ValueError("job_id must be a single project job name") return value def _validate_required_job_id(value: str) -> str: if not value.strip(): raise ValueError("job_id is required") return _validate_job_id(value) def _validate_prefix(value: str) -> str: if any(token in value for token in ("/", "\\", "..", "*", "?", "[", "]")): raise ValueError("prefix must not contain path or glob syntax") if value and not re.fullmatch( r"(?:Target\.)?[A-Za-z][A-Za-z0-9_-]{0,31}\.[A-Za-z][A-Za-z0-9_-]{0,63}", value, ): raise ValueError("prefix must be a canonical species.family identity") return value def _canonical_species_codes(values: Any) -> list[str]: if not isinstance(values, list): return [] return list( dict.fromkeys( value for value in values if isinstance(value, str) and re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,15}", value) ) ) def _normalize_result_stage(value: str) -> str: stage = str(value or "").strip() return { "differential_expression": "rnaseq_de", }.get(stage.lower(), stage) class InspectProjectContextArgs(BaseModel): model_config = ConfigDict(extra="forbid") job_id: str prefix: str = "" _job_id_is_local = field_validator("job_id")(_validate_required_job_id) _prefix_is_literal = field_validator("prefix")(_validate_prefix) class InspectAnalysisResultsArgs(BaseModel): model_config = ConfigDict(extra="forbid") job_id: str prefix: str = "" stage: str = "" scope_id: str = "" analysis_route: str = "" member_set_id: str = "" _job_id_is_local = field_validator("job_id")(_validate_required_job_id) _prefix_is_literal = field_validator("prefix")(_validate_prefix) _stage_is_canonical = field_validator("stage")(_normalize_result_stage) class QueryPlantAIRagArgs(BaseModel): model_config = ConfigDict(extra="forbid") question: str family: str = "" stage: str = "" class RecognizeAnalysisIntentArgs(BaseModel): model_config = ConfigDict(extra="forbid") query: str job_id: str = "" prefix: str = "" language: str = "zh" _job_id_is_local = field_validator("job_id")(_validate_job_id) _prefix_is_literal = field_validator("prefix")(_validate_prefix) class RecognizeUtilityIntentArgs(BaseModel): model_config = ConfigDict(extra="forbid") query: str class BoundInspectProjectContextArgs(BaseModel): model_config = ConfigDict(extra="forbid") prefix: str = "" _prefix_is_literal = field_validator("prefix")(_validate_prefix) class BoundInspectAnalysisResultsArgs(BaseModel): model_config = ConfigDict(extra="forbid") prefix: str = "" stage: str = "" scope_id: str = "" analysis_route: str = "" member_set_id: str = "" _prefix_is_literal = field_validator("prefix")(_validate_prefix) _stage_is_canonical = field_validator("stage")(_normalize_result_stage) class BoundRecognizeAnalysisIntentArgs(BaseModel): model_config = ConfigDict(extra="forbid") query: str prefix: str = "" language: str = "zh" _prefix_is_literal = field_validator("prefix")(_validate_prefix) class InspectJobFileArgs(BaseModel): model_config = ConfigDict(extra="forbid") path: str max_lines: int = Field(default=200, ge=1, le=500) class ListJobFilesArgs(BaseModel): model_config = ConfigDict(extra="forbid") roots: list[Literal["input", "final", "logs", "workspace"]] = Field( default_factory=lambda: ["input", "final", "logs"] ) limit: int = Field(default=100, ge=1, le=200) cursor: int = Field(default=0, ge=0) def normalize_rag_tool_result(raw: Any) -> dict[str, Any]: """Expose only truthful RAG availability, success, answer, and citations.""" data = raw if isinstance(raw, dict) else {} request_id = str(data.get("request_id") or f"rag-{uuid.uuid4().hex[:12]}") error_code = str(data.get("error_code") or "") available = error_code not in {"dual_kb_unavailable", "rag_unreachable"} answer = str(data.get("answer") or data.get("response") or "").strip() citations: list[dict[str, str]] = [] seen: set[tuple[str, str, str, str]] = set() raw_sources = data.get("sources") or data.get("citations") or [] if isinstance(raw_sources, list): for source in raw_sources: if not isinstance(source, dict): continue doc_id = str( source.get("doc_id") or source.get("paper_id") or source.get("pubmed_id") or source.get("id") or "" ).strip() title = str( source.get("title") or source.get("paper_title") or source.get("source") or doc_id or "" ).strip() locator = str( source.get("locator") or source.get("page") or source.get("section") or doc_id or source.get("target") or "" ).strip() if not title or not locator: continue url = str(source.get("url") or source.get("doi_url") or "").strip() if url: parsed = urllib.parse.urlsplit(url) if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc: url = "" key = (title, locator, url, doc_id) if key in seen: continue seen.add(key) citation = {"title": title, "locator": locator} if url: citation["url"] = url if doc_id: citation["doc_id"] = doc_id citations.append(citation) raw_success = data.get("success") is True if not raw_success: return { "registered": True, "available": available, "success": False, "request_id": request_id, "error_code": error_code or "rag_request_failed", "answer": "", "citations": [], } if not answer: error_code = "rag_empty_answer" elif not citations: error_code = "rag_no_usable_citations" if error_code: return { "registered": True, "available": available, "success": False, "request_id": request_id, "error_code": error_code, "answer": "", "citations": [], } return { "registered": True, "available": True, "success": True, "request_id": request_id, "error_code": "", "answer": answer, "citations": citations, } _FORBIDDEN_INTENT_KEYS = { "benchmarkMode", "forceRerun", "forceRerunStages", "scopeId", "scope_id", "memberSetId", "member_set_id", } def _without_forbidden_intent_keys(value: Any) -> Any: if isinstance(value, dict): return { key: _without_forbidden_intent_keys(item) for key, item in value.items() if key not in _FORBIDDEN_INTENT_KEYS } if isinstance(value, list): return [_without_forbidden_intent_keys(item) for item in value] return value def _manifest_identity(manifest: dict[str, Any]) -> dict[str, str]: def value(key: str) -> str: raw = manifest.get(key) or "" return raw if isinstance(raw, str) else "" return { "scope_id": value("scope_id"), "analysis_route": value("analysis_route"), "member_set_id": value("member_set_id"), } def _canonical_identity_is_complete(identity: dict[str, Any]) -> bool: return all( _safe_manifest_token(identity.get(key), max_length=512) for key in ("prefix", "scope_id", "analysis_route", "member_set_id") ) def _result_identity_is_complete(identity: dict[str, Any], *, stage: str) -> bool: """Validate identity at the requested stage's actual business boundary.""" if not all( _safe_manifest_token(identity.get(key), max_length=512) for key in ("prefix", "scope_id", "analysis_route") ): return False if stage == "identify": # Identify creates the candidate set; member_set_id exists only after # the user confirms those candidates. return identity.get("member_set_id") in {None, ""} return _safe_manifest_token(identity.get("member_set_id"), max_length=512) def _safe_manifest_token(value: Any, *, max_length: int) -> bool: allowed_punctuation = "._-:" return ( isinstance(value, str) and value == value.strip() and 0 < len(value) <= max_length and all( char.isascii() and (char.isalnum() or char in allowed_punctuation) for char in value ) and not Path(value).is_absolute() and "/" not in value and "\\" not in value and ".." not in value ) def _known_result_stage(stage: str) -> bool: return ( stage == "extension_stage" or stage_registry.get(stage) is not None or get_stage_capability(stage) is not None ) def _validated_manifest_rows( manifests: list[dict[str, Any]], *, final_dir: Path, ) -> list[dict[str, Any]]: try: resolved_final = final_dir.resolve(strict=True) except OSError: return [] validated = [] for row in manifests: path = Path(str(row.get("path") or "")) try: resolved_path = path.resolve(strict=True) resolved_path.relative_to(resolved_final) payload = json.loads(resolved_path.read_text(encoding="utf-8")) except (OSError, ValueError, TypeError): continue parameters = payload.get("parameters") if isinstance(payload.get("parameters"), dict) else {} raw = { "prefix": ( payload.get("prefix") or parameters.get("rnaseq_output_prefix") or parameters.get("prefix") ), "stage": payload.get("stage"), "scope_id": payload.get("scope_id"), "analysis_route": payload.get("analysis_route"), "member_set_id": payload.get("member_set_id"), "rnaseq_scope_id": ( payload.get("rnaseq_scope_id") or parameters.get("rnaseq_scope_id") ), } identity = {key: raw[key] for key in ("prefix", "scope_id", "analysis_route", "member_set_id")} is_rnaseq = str(raw["stage"] or "").startswith("rnaseq_") is_shared_prepare = raw["stage"] == "genome_prepare" if is_rnaseq: if not all( _safe_manifest_token(raw.get(key), max_length=512) for key in ("prefix", "rnaseq_scope_id") ): continue elif is_shared_prepare: if not _safe_manifest_token(raw.get("prefix"), max_length=512): continue elif not _result_identity_is_complete(identity, stage=str(raw["stage"] or "")): continue if not _safe_manifest_token(raw["stage"], max_length=128) or not _known_result_stage(raw["stage"]): continue validated.append({**row, **raw, "status": payload.get("status")}) return validated _UNSAFE_METRIC_KEY_TOKENS = { "path", "file", "dir", "directory", "warning", "diagnostic", "reason", "message", "raw", "error", "exception", "traceback", "log", "stdout", "stderr", "command", "detail", } _DROP_METRIC = object() class _ProjectionTooLarge(ValueError): pass def _safe_metric_key(key: Any, value: Any = None) -> bool: if not _safe_manifest_token(key, max_length=128): return False if isinstance(value, (bool, int, float)): return True tokens = {token for token in key.lower().replace("-", "_").split("_") if token} return not bool(tokens & _UNSAFE_METRIC_KEY_TOKENS) def _safe_metric_value(value: Any, *, depth: int = 0) -> Any: if isinstance(value, bool): return value if isinstance(value, int) and not isinstance(value, bool): return value if isinstance(value, float): return value if math.isfinite(value) else _DROP_METRIC if isinstance(value, str): return _DROP_METRIC if depth >= 4: return _DROP_METRIC if isinstance(value, list): if len(value) > 64: return _DROP_METRIC compact = [] for item in value: safe = _safe_metric_value(item, depth=depth + 1) if safe is not _DROP_METRIC: compact.append(safe) return compact if compact or not value else _DROP_METRIC if isinstance(value, dict): if len(value) > 128: return _DROP_METRIC compact = {} for key in sorted(value): if not _safe_metric_key(key, value[key]): continue safe = _safe_metric_value(value[key], depth=depth + 1) if safe is not _DROP_METRIC: compact[key] = safe return compact if compact or not value else _DROP_METRIC return _DROP_METRIC def _compact_stage_metrics(metrics_by_stage: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]: if len(metrics_by_stage) > 16: raise _ProjectionTooLarge compact: dict[str, dict[str, Any]] = {} for stage in sorted(metrics_by_stage): safe = _safe_metric_value(metrics_by_stage[stage]) if not isinstance(safe, dict): safe = {} compact[stage] = safe return compact def _compact_source_summary(summary: Any) -> dict[str, int]: if not isinstance(summary, dict): return {} allowed = { "manifest_count", "raw_evidence_count", "risk_count", "high_risk_count", "rag_reference_stage_count", "rag_reference_count", "claim_count", "major_claim_count", } return { key: value for key, value in sorted(summary.items()) if key in allowed and isinstance(value, int) and not isinstance(value, bool) and value >= 0 } def _compact_result_findings(report: dict[str, Any], *, limit: int = 3) -> list[dict[str, Any]]: findings = report.get("key_findings") if not isinstance(findings, list): return [] allowed = ("stage", "title", "description", "confidence", "interpretation_level") return [ _selected_fields(item, allowed) for item in findings[:limit] if isinstance(item, dict) ] def _compact_structured_interpretations(report: dict[str, Any]) -> dict[str, dict[str, Any]]: interpretations = report.get("structured_result_interpretations") if not isinstance(interpretations, dict): return {} allowed = ( "stage", "stage_name", "status", "result_interpretation", "limitations", "boundary", "interpretation_level", ) return { str(stage): _selected_fields(value, allowed) for stage, value in list(sorted(interpretations.items()))[:16] if _safe_binding_label(stage) and isinstance(value, dict) } def _bounded_tabular_artifact_rows( manifest_path: Path, *, final_dir: Path, limit: int = 5, ) -> list[dict[str, str]]: try: manifest = json.loads(manifest_path.read_text(encoding="utf-8")) except (OSError, ValueError, TypeError): return [] artifacts = manifest.get("artifacts") if not isinstance(artifacts, dict): return [] resolved_final = final_dir.resolve() excluded = {"plot", "raw", "manifest", "command", "environment", "checksum", "report"} candidates: list[tuple[str, Path]] = [] for key, artifact in sorted(artifacts.items()): if not isinstance(artifact, dict): continue raw_path = str(artifact.get("path") or "").strip() artifact_type = str(artifact.get("type") or "").strip().lower() suffix = Path(raw_path).suffix.lower() key_tokens = {token for token in str(key).lower().replace("-", "_").split("_") if token} if key_tokens & excluded: continue if artifact_type not in {"tsv", "csv"} and suffix not in {".tsv", ".txt", ".csv"}: continue declared = Path(raw_path) candidate = declared if declared.is_absolute() else ( final_dir.parent / declared if declared.parts[:1] == ("final",) else final_dir / declared ) try: resolved = candidate.resolve(strict=True) resolved.relative_to(resolved_final) except (OSError, ValueError): continue candidates.append((str(key), resolved)) if not candidates: return [] selected = candidates[0][1] try: with selected.open("r", encoding="utf-8", errors="replace", newline="") as handle: sample = handle.read(131_072) except OSError: return [] delimiter = "," if selected.suffix.lower() == ".csv" else "\t" reader = csv.DictReader(io.StringIO(sample), delimiter=delimiter) if not reader.fieldnames or len(reader.fieldnames) > 20: return [] rows: list[dict[str, str]] = [] for raw_row in reader: row = { str(key)[:128]: str(value or "")[:512] for key, value in list(raw_row.items())[:20] if key is not None } rows.append(row) if len(rows) >= limit: break return rows def _safe_binding_label(value: Any) -> bool: return _safe_manifest_token(value, max_length=128) def _candidate_identities( manifests: list[dict[str, Any]], *, prefix: str, stage: str, ) -> list[dict[str, str]]: identities: list[dict[str, str]] = [] seen: set[tuple[str, str, str, str]] = set() for manifest in manifests: manifest_prefix = str(manifest.get("prefix") or "") if prefix and manifest_prefix != prefix: continue if stage and str(manifest.get("stage") or "") != stage: continue identity = _manifest_identity(manifest) key = ( manifest_prefix, identity["scope_id"], identity["analysis_route"], identity["member_set_id"], ) if key in seen: continue seen.add(key) identities.append({"prefix": manifest_prefix, **identity}) return identities _INTEGRATED_ROUTE_PRIORITY = { "multi_species": 0, "single_species": 1, "confirmed_subfamily": 2, "subfamily_prediction": 3, "subfamily_evidence": 4, } def _manifest_row_metrics(row: dict[str, Any]) -> dict[str, Any]: metrics = row.get("metrics") if isinstance(metrics, dict) and metrics: return metrics try: payload = json.loads(Path(str(row.get("path") or "")).read_text(encoding="utf-8")) except (OSError, ValueError, TypeError): return {} metrics = payload.get("metrics") return metrics if isinstance(metrics, dict) else {} def _integrated_gene_family_rows( manifests: list[dict[str, Any]], *, prefix: str, ) -> tuple[dict[str, dict[str, Any]] | None, str]: """Select one evidence-bound successful Manifest per biological stage.""" rows_by_stage: dict[str, list[dict[str, Any]]] = {} for row in manifests: stage = str(row.get("stage") or "") if ( row.get("status") not in SUCCESS_STATUSES or stage.startswith("rnaseq_") or stage == "genome_prepare" or str(row.get("prefix") or "") != prefix ): continue rows_by_stage.setdefault(stage, []).append(row) selected: dict[str, dict[str, Any]] = {} for stage, rows in rows_by_stage.items(): rows_by_identity: dict[tuple[str, str, str], list[dict[str, Any]]] = {} for row in rows: identity = _manifest_identity(row) key = ( identity["scope_id"], identity["analysis_route"], identity["member_set_id"], ) rows_by_identity.setdefault(key, []).append(row) identity_rows: list[dict[str, Any]] = [] for identity_group in rows_by_identity.values(): metrics_by_row = [ (row, _manifest_row_metrics(row)) for row in identity_group ] for _, left in metrics_by_row: for _, right in metrics_by_row: shared_keys = set(left) & set(right) if any(left[key] != right[key] for key in shared_keys): return None, stage identity_rows.append( sorted( metrics_by_row, key=lambda item: ( -len(item[1]), str(item[0].get("path") or ""), ), )[0][0] ) best_priority = min( _INTEGRATED_ROUTE_PRIORITY.get(str(row.get("analysis_route") or ""), 50) for row in identity_rows ) best_rows = [ row for row in identity_rows if _INTEGRATED_ROUTE_PRIORITY.get( str(row.get("analysis_route") or ""), 50 ) == best_priority ] if len(best_rows) != 1: return None, stage selected[stage] = best_rows[0] return selected, "" def _selected_fields(item: dict[str, Any], fields: tuple[str, ...]) -> dict[str, Any]: return {key: item[key] for key in fields if item.get(key) not in (None, "", [], {})} class _NoCostLLMManager: def __init__(self, wrapped: LLMManager): self._wrapped = wrapped def __getattr__(self, name: str) -> Any: return getattr(self._wrapped, name) def call(self, *args: Any, **kwargs: Any) -> Any: kwargs["track_cost"] = False return self._wrapped.call(*args, **kwargs) async def acall(self, *args: Any, **kwargs: Any) -> Any: kwargs["track_cost"] = False return await self._wrapped.acall(*args, **kwargs) def _matching_manifest_rows( manifests: list[dict[str, Any]], *, prefix: str, stage: str, identity: dict[str, str], ) -> list[dict[str, Any]]: return [ manifest for manifest in manifests if (not prefix or str(manifest.get("prefix") or "") == prefix) and (not stage or str(manifest.get("stage") or "") == stage) and _manifest_identity(manifest) == identity ] def _bind_exact_manifest_sources( manifests: list[dict[str, Any]], *, identity: dict[str, str], final_dir: Path, ) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]] | None: resolved_final = final_dir.resolve() rows_by_stage: dict[str, list[dict[str, Any]]] = {} for manifest in manifests: rows_by_stage.setdefault(str(manifest.get("stage") or ""), []).append(manifest) metrics_by_stage: dict[str, dict[str, Any]] = {} bindings: dict[str, dict[str, Any]] = {} for stage, rows in rows_by_stage.items(): if not _safe_binding_label(stage): return None payloads: list[tuple[Path, dict[str, Any]]] = [] for row in rows: path = Path(str(row.get("path") or "")) try: resolved_path = path.resolve() resolved_path.relative_to(resolved_final) payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError, TypeError): return None if _manifest_identity(payload) != identity or str(payload.get("stage") or "") != stage: return None payloads.append((path, payload)) metric_variants = {json.dumps(payload.get("metrics") or {}, sort_keys=True) for _, payload in payloads} if len(metric_variants) != 1: return None path, payload = sorted(payloads, key=lambda item: str(item[0]))[0] if payload.get("status") not in SUCCESS_STATUSES: return None metrics_by_stage[stage] = payload.get("metrics") or {} bindings[stage] = { "manifest_ref": (Path("final") / path.resolve().relative_to(resolved_final)).as_posix(), "canonical_identity": dict(identity), "stage": stage, "status": payload.get("status", ""), } return metrics_by_stage, bindings def _report_identity_is_compatible(metadata: dict[str, Any], identity: dict[str, str]) -> bool: report_scope = str(metadata.get("scope_id") or "") report_route = str(metadata.get("analysis_route") or "") report_member = str(metadata.get("member_set_id") or "") if report_scope != identity["scope_id"]: return False if report_route == identity["analysis_route"]: return report_member == identity["member_set_id"] normalized_target_plus_reference = ( identity["analysis_route"] == "multi_species" and "target_plus_reference" in identity["scope_id"] and report_route == "target_plus_reference" ) return normalized_target_plus_reference and report_member in {"", identity["member_set_id"]} def build_plantai_conversation_tools( *, llm_manager: LLMManager, project_root: Path, bound_job_id: str = "", goal_ledger: GoalLedger | None = None, allow_legacy_manifest_migration: bool = True, progress_callback: Callable[[dict[str, Any]], None] | None = None, ) -> ConversationToolRegistry: project_root = Path(project_root) bound_job_id = _validate_job_id(str(bound_job_id or "")) registry = ConversationToolRegistry(progress_callback=progress_callback) def current_job(args: BaseModel) -> str: return bound_job_id or str(getattr(args, "job_id", "") or "") context_options = ( {} if allow_legacy_manifest_migration else {"allow_legacy_manifest_migration": False} ) async def inspect_project(args: InspectProjectContextArgs) -> dict[str, Any]: job_id = current_job(args) index = await asyncio.to_thread( build_project_context, job_id, args.prefix, project_root, **context_options, ) data = index.to_dict() scopes = [ _selected_fields( item, ( "scope_id", "scope_mode", "family", "target_species", "reference_species", "comparative_species", "selected_species", ), ) for item in data.get("scope_files", []) ] for item in scopes: for key in ( "target_species", "reference_species", "comparative_species", "selected_species", ): if key in item: item[key] = _canonical_species_codes(item[key]) manifests = [ _selected_fields( item, ( "stage", "status", "prefix", "scope_id", "analysis_route", "member_set_id", "scope_mode", "family", "target_species", "reference_species", "comparative_species", ), ) for item in data.get("manifests", []) ] for item in manifests: for key in ( "target_species", "reference_species", "comparative_species", ): if key in item: item[key] = _canonical_species_codes(item[key]) stage_status = { stage: _selected_fields(status, ("stage", "status")) for stage, status in data.get("stage_status", {}).items() } warnings = [ _selected_fields(item, ("code", "message", "stage")) for item in data.get("warnings", []) ] return { "job_id": data.get("job_id", job_id), "family": data.get("family", ""), "species_roles": { "target": _canonical_species_codes(data.get("target_species", [])), "reference": _canonical_species_codes(data.get("reference_species", [])), "comparative": _canonical_species_codes(data.get("comparative_species", [])), }, "scopes": scopes, "manifests": manifests, "stage_status": stage_status, "warnings": warnings, } async def inspect_results(args: InspectAnalysisResultsArgs) -> dict[str, Any]: job_id = current_job(args) index = await asyncio.to_thread( build_project_context, job_id, args.prefix, project_root, **context_options, ) index_data = index.to_dict() final_dir = project_root / "workspace" / job_id / "final" manifests = _validated_manifest_rows( index_data.get("manifests", []), final_dir=final_dir, ) rnaseq_rows = [ row for row in manifests if str(row.get("stage") or "").startswith("rnaseq_") and row.get("status") in SUCCESS_STATUSES and (not args.stage or row.get("stage") == args.stage) and (not args.prefix or row.get("prefix") == args.prefix) ] if args.stage == "genome_prepare": prepare_rows = [ row for row in manifests if row.get("stage") == "genome_prepare" and row.get("status") in SUCCESS_STATUSES ] if len(prepare_rows) != 1: return { "success": False, "error_code": ( "ambiguous_evidence_binding" if prepare_rows else "result_not_found" ), "message": "Genome preparation result is missing or ambiguous.", } selected = prepare_rows[0] effective_prefix = str(selected.get("prefix") or "genome_prepare") report = await asyncio.to_thread( build_evidence_report, final_dir, effective_prefix, job_id=job_id, include_stages=["genome_prepare"], ) if report is None: return { "success": False, "error_code": "result_not_found", "message": "No analysis result matched the request.", } manifest_path = Path(str(selected.get("path") or "")) return { "projection_version": "compact_v1", "success": True, "canonical_identity": {"prefix": effective_prefix}, "metadata": {"prefix": effective_prefix}, "result_agent": { "stage_metrics": _compact_stage_metrics( {"genome_prepare": dict(selected.get("metrics") or {})} ), "key_findings": _compact_result_findings(report), "structured_result_interpretations": _compact_structured_interpretations(report), "top_results": {}, }, "source_bindings": { "genome_prepare": { "manifest_ref": ( Path("final") / manifest_path.resolve().relative_to(final_dir.resolve()) ).as_posix(), "stage": "genome_prepare", "status": selected.get("status", ""), } }, "source_summary": _compact_source_summary(report.get("source_summary")), } if not args.stage and rnaseq_rows: rnaseq_identities = sorted( { ( str(row.get("prefix") or ""), str(row.get("rnaseq_scope_id") or ""), ) for row in rnaseq_rows if str(row.get("prefix") or "").strip() and str(row.get("rnaseq_scope_id") or "").strip() } ) if len(rnaseq_identities) != 1: return { "success": False, "error_code": "ambiguous_scope", "message": "Multiple RNA-seq result identities match; provide prefix or stage.", "candidates": [ {"prefix": prefix, "rnaseq_scope_id": scope_id} for prefix, scope_id in rnaseq_identities ], } effective_prefix, rnaseq_scope_id = rnaseq_identities[0] selected_rows = [ row for row in rnaseq_rows if str(row.get("prefix") or "") == effective_prefix and str(row.get("rnaseq_scope_id") or "") == rnaseq_scope_id ] rows_by_stage: dict[str, list[dict[str, Any]]] = {} for row in selected_rows: rows_by_stage.setdefault(str(row.get("stage") or ""), []).append(row) canonical_rows: dict[str, dict[str, Any]] = {} for stage, rows in rows_by_stage.items(): metric_variants = { json.dumps(row.get("metrics") or {}, sort_keys=True) for row in rows } if not stage or len(metric_variants) != 1: return { "success": False, "error_code": "ambiguous_evidence_binding", "message": "RNA-seq manifests disagree within one stage identity.", } canonical_rows[stage] = sorted( rows, key=lambda row: str(row.get("path") or "") )[0] report = await asyncio.to_thread( build_evidence_report, final_dir, effective_prefix, job_id=job_id, rnaseq_scope_id=rnaseq_scope_id, include_stages=list(canonical_rows), ) if report is None: return { "success": False, "error_code": "result_not_found", "message": "No analysis result matched the request.", } top_results: dict[str, list[dict[str, str]]] = {} source_bindings: dict[str, dict[str, Any]] = {} stage_metrics: dict[str, dict[str, Any]] = {} for stage, row in canonical_rows.items(): manifest_path = Path(str(row.get("path") or "")) stage_metrics[stage] = dict(row.get("metrics") or {}) if stage in { "rnaseq_de", "rnaseq_go_enrich", "rnaseq_kegg_enrich", }: rows = await asyncio.to_thread( _bounded_tabular_artifact_rows, manifest_path, final_dir=final_dir, ) if rows: top_results[stage] = rows source_bindings[stage] = { "manifest_ref": ( Path("final") / manifest_path.resolve().relative_to(final_dir.resolve()) ).as_posix(), "stage": stage, "status": row.get("status", ""), } projection = { "projection_version": "compact_v1", "success": True, "canonical_identity": { "prefix": effective_prefix, "rnaseq_scope_id": rnaseq_scope_id, }, "metadata": { "prefix": effective_prefix, "rnaseq_scope_id": rnaseq_scope_id, }, "result_agent": { "stage_metrics": _compact_stage_metrics(stage_metrics), "key_findings": _compact_result_findings(report, limit=16), "structured_result_interpretations": _compact_structured_interpretations(report), "top_results": top_results, }, "source_bindings": source_bindings, "source_summary": _compact_source_summary(report.get("source_summary")), } if len(json.dumps(projection, ensure_ascii=False, sort_keys=True)) > 20_000: projection["result_agent"]["stage_metrics"] = {} if len(json.dumps(projection, ensure_ascii=False, sort_keys=True)) > 20_000: return { "success": False, "error_code": "result_payload_too_large", "message": "The compact result exceeds the safe response budget.", } return projection has_exact_identity_selector = any( (args.scope_id, args.analysis_route, args.member_set_id) ) if not args.stage and not has_exact_identity_selector: effective_prefix = args.prefix if not effective_prefix: family = str(index_data.get("family") or "").strip() target_species = index_data.get("target_species") or [] expected_prefix = ( f"{target_species[0]}.{family}" if len(target_species) == 1 and family else "" ) available_prefixes = { str(row.get("prefix") or "") for row in manifests if row.get("status") in SUCCESS_STATUSES and not str(row.get("stage") or "").startswith("rnaseq_") and str(row.get("stage") or "") != "genome_prepare" } if expected_prefix in available_prefixes: effective_prefix = expected_prefix else: canonical_prefixes = { value for value in available_prefixes if re.fullmatch( r"[A-Za-z][A-Za-z0-9_-]{0,31}\.[A-Za-z][A-Za-z0-9_-]{0,63}", value, ) } if len(canonical_prefixes) == 1: effective_prefix = next(iter(canonical_prefixes)) elif len(canonical_prefixes) > 1: return { "success": False, "error_code": "ambiguous_prefix", "message": "Multiple gene-family result prefixes match; provide prefix.", "candidates": sorted(canonical_prefixes), } integrated_identities = [ candidate for candidate in _candidate_identities( manifests, prefix=effective_prefix, stage="", ) if _canonical_identity_is_complete(candidate) ] if effective_prefix and len(integrated_identities) > 1: integrated_manifests = list(manifests) ambiguous_stages: list[str] = [] while True: canonical_rows, ambiguous_stage = _integrated_gene_family_rows( integrated_manifests, prefix=effective_prefix, ) if canonical_rows is not None or not ambiguous_stage: break ambiguous_stages.append(ambiguous_stage) integrated_manifests = [ row for row in integrated_manifests if str(row.get("stage") or "") != ambiguous_stage ] if canonical_rows is None or (not canonical_rows and ambiguous_stages): return { "success": False, "error_code": "ambiguous_evidence_binding", "message": "Multiple equally valid result identities match one stage.", "stage": ambiguous_stages[0] if ambiguous_stages else ambiguous_stage, } if canonical_rows: source_bindings: dict[str, dict[str, Any]] = {} stage_metrics: dict[str, dict[str, Any]] = {} key_findings: list[dict[str, Any]] = [] structured_interpretations: dict[str, dict[str, Any]] = {} source_summary: dict[str, int] = {} for stage, row in sorted(canonical_rows.items()): identity = _manifest_identity(row) report = await asyncio.to_thread( build_evidence_report, final_dir, effective_prefix, job_id=job_id, scope_id=identity["scope_id"], analysis_route=identity["analysis_route"], member_set_id=identity["member_set_id"], include_stages=[stage], ) if isinstance(report, dict): key_findings.extend(_compact_result_findings(report, limit=1)) structured_interpretations.update( _compact_structured_interpretations(report) ) for key, value in _compact_source_summary( report.get("source_summary") ).items(): source_summary[key] = source_summary.get(key, 0) + value manifest_path = Path(str(row.get("path") or "")) stage_metrics[stage] = dict(_manifest_row_metrics(row)) source_bindings[stage] = { "manifest_ref": ( Path("final") / manifest_path.resolve().relative_to(final_dir.resolve()) ).as_posix(), "stage": stage, "status": row.get("status", ""), "canonical_identity": identity, } projection = { "projection_version": "compact_v1", "success": True, "canonical_identity": {"prefix": effective_prefix}, "metadata": { "prefix": effective_prefix, "report_mode": "integrated_gene_family", }, "result_agent": { "completed_stages": sorted(canonical_rows), "ambiguous_stages": [ { "stage": stage, "reason": "multiple_canonical_identities", } for stage in ambiguous_stages ], "stage_metrics": _compact_stage_metrics(stage_metrics), "key_findings": key_findings[:16], "structured_result_interpretations": structured_interpretations, }, "source_bindings": source_bindings, "source_summary": source_summary, } if len(json.dumps(projection, ensure_ascii=False, sort_keys=True)) > 20_000: projection["result_agent"]["stage_metrics"] = {} if len(json.dumps(projection, ensure_ascii=False, sort_keys=True)) > 20_000: return { "success": False, "error_code": "result_payload_too_large", "message": "The compact result exceeds the safe response budget.", } return projection if args.stage.startswith("rnaseq_"): if len(rnaseq_rows) > 1: return { "success": False, "error_code": "ambiguous_scope", "message": "Multiple RNA-seq results match; provide prefix.", "candidates": [ { "prefix": row["prefix"], "rnaseq_scope_id": row["rnaseq_scope_id"], "stage": row["stage"], } for row in rnaseq_rows ], } if not rnaseq_rows: return { "success": False, "error_code": "result_not_found", "message": "No analysis result matched the request.", } selected = rnaseq_rows[0] effective_prefix = str(selected["prefix"]) rnaseq_scope_id = str(selected["rnaseq_scope_id"]) report = await asyncio.to_thread( build_evidence_report, final_dir, effective_prefix, job_id=job_id, rnaseq_scope_id=rnaseq_scope_id, include_stages=[str(selected["stage"])], ) if report is None: return { "success": False, "error_code": "result_not_found", "message": "No analysis result matched the request.", } metrics = dict(selected.get("metrics") or {}) manifest_path = Path(str(selected["path"])) if not metrics: metrics = json.loads(manifest_path.read_text(encoding="utf-8")).get("metrics") or {} stage = str(selected["stage"]) top_rows = await asyncio.to_thread( _bounded_tabular_artifact_rows, manifest_path, final_dir=final_dir, ) return { "success": True, "canonical_identity": { "prefix": effective_prefix, "rnaseq_scope_id": rnaseq_scope_id, }, "metadata": { "prefix": effective_prefix, "rnaseq_scope_id": rnaseq_scope_id, }, "result_agent": { "stage_metrics": _compact_stage_metrics({stage: metrics}), "key_findings": _compact_result_findings(report), "structured_result_interpretations": _compact_structured_interpretations(report), "top_results": {stage: top_rows} if top_rows else {}, }, "source_bindings": { stage: { "manifest_ref": ( Path("final") / Path(str(selected["path"])).resolve().relative_to(final_dir.resolve()) ).as_posix(), "stage": selected["stage"], "status": selected["status"], } }, "source_summary": _compact_source_summary(report.get("source_summary")), } candidates = _candidate_identities(manifests, prefix=args.prefix, stage=args.stage) candidates = [ candidate for candidate in candidates if _result_identity_is_complete(candidate, stage=args.stage) ] requested = { "scope_id": args.scope_id, "analysis_route": args.analysis_route, "member_set_id": args.member_set_id, } matching = [ candidate for candidate in candidates if all(not requested[key] or candidate[key] == requested[key] for key in requested) ] if len(matching) > 1: canonical_identities = { (candidate["scope_id"], candidate["analysis_route"], candidate["member_set_id"]) for candidate in matching } prefixes = {candidate["prefix"] for candidate in matching} ambiguous_prefix = len(canonical_identities) == 1 and len(prefixes) > 1 return { "success": False, "error_code": "ambiguous_prefix" if ambiguous_prefix else "ambiguous_scope", "message": ( "Multiple result prefixes match; provide prefix." if ambiguous_prefix else "Multiple result identities match; provide scope_id, analysis_route, and member_set_id." ), "candidates": matching, } if candidates and not matching: return { "success": False, "error_code": "scope_not_found", "message": "No result matches the requested identity.", "candidates": candidates, } if not matching: return { "success": False, "error_code": "result_not_found", "message": "No analysis result matched the request.", } selected = matching[0] effective_prefix = selected["prefix"] identity = {key: selected[key] for key in requested} same_scope = [ {key: candidate[key] for key in requested} for candidate in candidates if candidate["scope_id"] == identity["scope_id"] ] if any(candidate != identity for candidate in same_scope): return { "success": False, "error_code": "ambiguous_evidence_binding", "message": "Existing evidence loading cannot isolate this member set without mixing identities.", "candidates": same_scope, } source_rows = _matching_manifest_rows( manifests, prefix=effective_prefix, stage=args.stage, identity=identity, ) bound_sources = await asyncio.to_thread( _bind_exact_manifest_sources, source_rows, identity=identity, final_dir=final_dir, ) if bound_sources is None: return { "success": False, "error_code": "ambiguous_evidence_binding", "message": "Exact manifest sources are missing, inconsistent, or disagree on metrics.", "canonical_identity": identity, } bound_metrics, source_bindings = bound_sources report = await asyncio.to_thread( build_evidence_report, final_dir, effective_prefix, job_id=job_id, scope_id=identity["scope_id"], analysis_route=identity["analysis_route"], member_set_id=identity["member_set_id"], include_stages=[args.stage] if args.stage else None, ) if report is None: return { "success": False, "error_code": "result_not_found", "message": "No analysis result matched the request.", "canonical_identity": identity, } report_metadata = report.get("metadata") if isinstance(report.get("metadata"), dict) else {} if not _report_identity_is_compatible(report_metadata, identity): return { "success": False, "error_code": "ambiguous_evidence_binding", "message": "The evidence report identity is incompatible with the selected Manifest identity.", "canonical_identity": identity, } try: projection = { "projection_version": "compact_v1", "success": True, "canonical_identity": identity, "metadata": {"prefix": effective_prefix, **identity}, "result_agent": { "stage_metrics": _compact_stage_metrics(bound_metrics), "key_findings": _compact_result_findings(report), "structured_result_interpretations": _compact_structured_interpretations(report), }, "source_bindings": source_bindings, "source_summary": _compact_source_summary(report.get("source_summary")), } except _ProjectionTooLarge: projection = None if ( projection is None or len(json.dumps(projection, ensure_ascii=False, sort_keys=True)) > 20_000 ): return { "success": False, "error_code": "result_payload_too_large", "message": "The compact result exceeds the safe response budget.", } return projection async def query_rag(args: QueryPlantAIRagArgs) -> dict[str, Any]: try: raw = await asyncio.to_thread(query_dual_kb_qa, args.question, top_k=10) except Exception: raw = { "success": False, "error_code": "dual_kb_unavailable", "answer": "", "sources": [], } return normalize_rag_tool_result(raw) def intent_query_with_confirmed_context( query: str, *, workflow_id: str | None = None, guidance: bool = False, language: str = "zh", ) -> str: """Adapt MainAgent memory to the original IntentAgent text contract.""" text = str(query or "").strip() en = language == "en" if ( text.startswith("执行数据准备（genome_prepare）") and "当前业务动作已由数据准备契约唯一确定" in text ): return text if goal_ledger is None: return text terms = goal_ledger.user_terms context: list[str] = [] family = str(terms.get("gene_family") or "").strip() if family: context.append(f"Gene family: {family}" if en else f"基因家族 {family}") for key, label_zh, label_en in ( ("target_species", "目标物种", "Target species"), ("reference_species", "参考物种", "Reference species"), ("comparative_species", "比较物种", "Comparative species"), ): label = label_en if en else label_zh values = terms.get(key) or () if isinstance(values, (list, tuple)): cleaned = [ str(value).strip() for value in values if str(value or "").strip() ] if cleaned: context.append(f"{label} {','.join(cleaned)}") rna_route = str(terms.get("rna_route") or "").strip().lower() if workflow_id in {"rnaseq_whole_genome", "rnaseq_family_member"}: business_context = load_rnaseq_business_context(workflow_id) mode = business_context["analysis_mode"]["mode"] experiment_design = business_context["authoritative_inputs"][ "experiment_design" ] if mode == "whole_genome": enabled_designs = "、".join(business_context["enabled_designs"]) disabled_designs = "、".join(business_context["disabled_designs"]) if en: context.append( f"RNA-seq mode: whole genome; {experiment_design} is the authoritative experimental design input; " f"interpretable designs: {enabled_designs}, {disabled_designs} disabled this round; " "design or batch downgrades must not happen silently" ) else: context.append( f"RNA-seq业务模式 全基因组；{experiment_design} 是实验设计权威输入；" f"可解释设计为 {enabled_designs}，{disabled_designs} 本轮禁用；" "设计或 batch 降级不得静默发生" ) else: if en: context.append( "RNA-seq mode: gene family member; use successful whole-genome differential expression results " "and precisely confirmed gene family members for projection; must not auto-select the first member file, " "must not implicitly re-run differential expression" ) else: context.append( "RNA-seq业务模式 基因家族成员；使用已成功的全基因组差异表达结果" "和精确确认的基因家族成员进行投影；不得自动选择首个成员文件，" "不得隐式重跑差异表达" ) elif rna_route == "whole_genome": context.append( "RNA-seq mode: whole genome" if en else "RNA-seq模式 全基因组" ) elif rna_route == "family_member": context.append( "RNA-seq mode: gene family member" if en else "RNA-seq模式 基因家族成员" ) job_id = bound_job_id or str(terms.get("active_job_id") or "").strip() formal_config_exists = bool( job_id and (project_root / "input" / job_id / "config.xlsx").is_file() ) config_confirmed = ( bool(terms.get("config_confirmed")) or "config.xlsx 已确认" in goal_ledger.approved_decisions or formal_config_exists ) subfamily_confirm_observed = False if job_id: final_dir = project_root / "workspace" / job_id / "final" for manifest_path in final_dir.glob("*.manifest.json") if final_dir.exists() else (): try: manifest = json.loads(manifest_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): continue stage = str(manifest.get("stage") or "").strip() status = manifest.get("status") manifest_prefix = str(manifest.get("prefix") or "").strip() family_matches = not family or bool( re.search( rf"(?:^|[.]){re.escape(family)}(?:[.]|$)", manifest_prefix, flags=re.IGNORECASE, ) ) if ( stage == "subfamily_confirm" and status == "success" and family_matches ): subfamily_confirm_observed = True genome_prepare_observed = False canonical_request = None if config_confirmed and not genome_prepare_observed: if en: context.append( "Formal config.xlsx confirmed; this indicates configuration has been verified, " "not that data preparation is complete. No data preparation completion evidence in current job; " "specific analysis intent and stages still determined by the existing analysis system" ) else: context.append( "正式 config.xlsx 已确认；这只表示配置已核对，不等于数据准备已经完成，" "当前 job 尚无数据准备完成证据；具体分析意图和阶段仍由现有分析系统判断" ) if guidance and workflow_id in { "data_preparation", "gene_family", "rnaseq_whole_genome", "rnaseq_family_member", }: contract = load_canonical_workflow_contract("data_preparation") transition = next( item for item in contract.transitions if item.from_state_id == "config_confirmed" ) actions = {item.id: item for item in contract.actions} target_stage = actions[transition.action_id].target_stage if en: canonical_request = ( f"Execute data preparation ({target_stage})\n" "Business action uniquely determined by data preparation contract" ) else: canonical_request = ( f"执行数据准备（{target_stage}）\n" "当前业务动作已由数据准备契约唯一确定" ) if family and subfamily_confirm_observed: if en: context.append( "Current job's subfamily_confirm completed successfully; " "family members and subfamily classification table confirmed. " "This is a current project state fact; specific analysis intent and stages still determined by the existing analysis system" ) else: context.append( "当前 job 的 subfamily_confirm 已成功完成，家族成员和亚家族分类表已确认，" "这是当前项目状态事实；具体分析意图和阶段仍由现有分析系统判断" ) if canonical_request: return canonical_request if not context: return text if en: return f"{text}\nConfirmed context: {'; '.join(context)}." return f"{text}\n已确认上下文：{'；'.join(context)}。" registry.set_analysis_handoff_resolver(intent_query_with_confirmed_context) async def recognize_intent(args: RecognizeAnalysisIntentArgs) -> dict[str, Any]: job_id = current_job(args) state = AgentState( session_id=f"router:{job_id or 'unbound'}", job_id=job_id, results={"completed_stages": []}, ) message = AgentMessage( sender="router", receiver="intent_v2", message_type="recognize_intent", payload={ "query": intent_query_with_confirmed_context(args.query, language=args.language), "job_id": job_id, "prefix": args.prefix, "workspace": str(project_root / "workspace" / job_id) if job_id else "", **( { "goal_context": { "active_goal": goal_ledger.active_goal, "user_terms": goal_ledger.model_dump( mode="json", round_trip=True )["user_terms"], } } if goal_ledger is not None else {} ), }, ) returned_state = await IntentAgentV2(_NoCostLLMManager(llm_manager)).execute(state, message) candidate = dict( ((returned_state.results.get("results") or {}).get("intent") or {}) ) stages = [str(stage) for stage in candidate.get("stages") or ()] if any(stage.startswith("rnaseq_") for stage in stages): query_lower = str(args.query or "").lower() family_member = any( token in query_lower for token in ( "基因家族成员", "家族成员", "family member", "family-member", ) ) whole_genome = any( token in query_lower for token in ("全基因组", "whole genome", "whole-genome") ) if family_member or whole_genome: rna_route = "family_member" if family_member else "whole_genome" foreground_mode = "family" if family_member else "global" candidate["rna_route"] = rna_route candidate["foreground_mode"] = foreground_mode params = dict(candidate.get("params") or {}) params["analysisRoute"] = rna_route params["rnaseqForegroundMode"] = foreground_mode candidate["params"] = params return _without_forbidden_intent_keys(candidate) def recognize_utility(args: RecognizeUtilityIntentArgs) -> dict[str, Any]: from gfa_core.mas.utils.intent_parser import parse_sequence_extract_request params = parse_sequence_extract_request(args.query) if not params: return { "success": False, "error_code": "utility_intent_not_found", "message": "No supported utility request was recognized.", } return { "success": True, "taskType": "utility", "stage": "sequence_extract_by_id", "stages": ["sequence_extract_by_id"], "utility_tool": ( "fasta_header_clean" if params.get("sequenceExtractCleanHeaders") else "sequence_extract_by_id" ), "params": params, } def inspect_job_file(args: InspectJobFileArgs) -> dict[str, Any]: from gfa_core.mas.job_file_reader import read_job_file_preview return read_job_file_preview( project_root, bound_job_id, args.path, max_lines=args.max_lines, ) def list_current_job_files(args: ListJobFilesArgs) -> dict[str, Any]: from gfa_core.mas.job_file_reader import list_job_files return list_job_files( project_root, bound_job_id, roots=tuple(args.roots), limit=args.limit, cursor=args.cursor, ) registry.register( name="inspect_project_context", description="Inspect one PlantAI job's project context without changing it.", args_model=(BoundInspectProjectContextArgs if bound_job_id else InspectProjectContextArgs), handler=inspect_project, permission="read_only", ) registry.register( name="inspect_analysis_results", description="Read evidence from one canonical PlantAI analysis result identity.", args_model=(BoundInspectAnalysisResultsArgs if bound_job_id else InspectAnalysisResultsArgs), handler=inspect_results, permission="read_only", ) registry.register( name="query_plantai_rag", description="Query PlantAI's existing gene-family knowledge service.", args_model=QueryPlantAIRagArgs, handler=query_rag, permission="read_only", ) registry.register( name="recognize_analysis_intent", description="Ask IntentAgentV2 for a candidate analysis intent without planning or execution.", args_model=(BoundRecognizeAnalysisIntentArgs if bound_job_id else RecognizeAnalysisIntentArgs), handler=recognize_intent, permission="pure", ) registry.register( name="recognize_utility_intent", description="Recognize one supported utility request without planning or execution.", args_model=RecognizeUtilityIntentArgs, handler=recognize_utility, permission="pure", ) if bound_job_id: registry.register( name="list_job_files", description=( "List real safe file handles in the current PlantAI job. " "Use input/final/logs by default; request workspace only when the user asks for intermediates or all files." ), args_model=ListJobFilesArgs, handler=list_current_job_files, permission="read_only", max_result_chars=70_000, ) registry.register( name="inspect_job_file", description="Preview one safe text file attached to the current PlantAI job.", args_model=InspectJobFileArgs, handler=inspect_job_file, permission="read_only", max_result_chars=70_000, ) return registry