diff --git a/TODO.md b/TODO.md index e9d3668..5186326 100644 --- a/TODO.md +++ b/TODO.md @@ -128,18 +128,44 @@ --- -### [TODO][P1] 增加 MCP 结果读取接口 `get_delivery_payload` +### [DONE][P1] 增加 MCP 结果读取接口 `get_delivery_payload` 目标： - 按 run_id 读取 delivery payload +完成情况： +- 已通过 MCP 暴露 `get_delivery_payload` +- 查询优先复用 `run-state.json` 已注册 artifacts，其次回退标准产物路径、`run-report.json` 引用和 run 目录扫描 +- 返回补充了 `artifact`、`payload_schema_version`、`generated_at`、`delivery_date`、`candidate_count`、`stats`，并保留完整 `payload` + +改动文件： +- `src/summary_mcp/runtime/query_service.py` +- `src/summary_mcp/server.py` +- `src/summary_mcp/runtime/__init__.py` + +遗留风险： +- 历史 run 的 payload 若既未注册也不在标准路径下，只能依赖 `run-report.json` 引用或目录扫描做兼容发现 + --- -### [TODO][P1] 增加 MCP 结果读取接口 `get_run_report` +### [DONE][P1] 增加 MCP 结果读取接口 `get_run_report` 目标： - 按 run_id 读取 run report +完成情况： +- 已通过 MCP 暴露 `get_run_report` +- 查询优先复用 `run-state.json` 已注册 artifacts，其次回退标准产物路径、历史 `run-report.json` 固定位置和 run 目录扫描 +- 返回补充了 `artifact`、核心计数摘要、规范化后的关键产物路径与 `keyword_index`，并保留完整 `report` + +改动文件： +- `src/summary_mcp/runtime/query_service.py` +- `src/summary_mcp/runtime/__init__.py` +- `src/summary_mcp/server.py` + +遗留风险： +- 历史 run 的 `run-report.json` 内嵌路径仍保留原始绝对路径于 `report` 字段，当前仅在顶层摘要字段做规范化，避免改变历史文件真相 + --- ### [TODO][P2] 设计并实现 `resume_run` diff --git a/src/summary_mcp/runtime/__init__.py b/src/summary_mcp/runtime/__init__.py index f32c948..33f2c2e 100644 --- a/src/summary_mcp/runtime/__init__.py +++ b/src/summary_mcp/runtime/__init__.py @@ -1,9 +1,11 @@ -from .query_service import get_run_status, list_run_artifacts, list_runs +from .query_service import get_delivery_payload, get_run_report, get_run_status, list_run_artifacts, list_runs from .run_store import RunStore from .state_models import ArtifactRecord, RecoveryState, RunError, RunState, StageState __all__ = [ "ArtifactRecord", + "get_delivery_payload", + "get_run_report", "RecoveryState", "RunError", "RunState", diff --git a/src/summary_mcp/runtime/query_service.py b/src/summary_mcp/runtime/query_service.py index a544807..8cd8f11 100644 --- a/src/summary_mcp/runtime/query_service.py +++ b/src/summary_mcp/runtime/query_service.py @@ -98,6 +98,72 @@ def list_run_artifacts(*, run_id: str) -> dict[str, Any]: } +def get_delivery_payload(*, run_id: str) -> dict[str, Any]: + record = _resolve_run_record(run_id) + artifact = _resolve_artifact( + record, + artifact_name="delivery_payload", + relative_path=Path("candidates/openclaw-delivery-payload.json"), + fallback_stage="build_delivery_payload", + fallback_kind="json", + report_field="delivery_output", + ) + payload = _load_json(_resolve_repo_path(artifact["path"])) + candidates = payload.get("candidates") + candidate_count = len(candidates) if isinstance(candidates, list) else 0 + stats = payload.get("stats") if isinstance(payload.get("stats"), dict) else None + response = _build_run_lookup_response(record) + response.update( + { + "artifact": artifact, + "payload_schema_version": payload.get("schema_version"), + "generated_at": payload.get("generated_at"), + "delivery_date": payload.get("date"), + "candidate_count": candidate_count, + "stats": stats, + "payload": payload, + } + ) + return response + + +def get_run_report(*, run_id: str) -> dict[str, Any]: + record = _resolve_run_record(run_id) + artifact = _resolve_artifact( + record, + artifact_name="run_report", + relative_path=Path("run-report.json"), + fallback_stage="write_run_report", + fallback_kind="json", + report_field="report_output", + ) + report = _load_json(_resolve_repo_path(artifact["path"])) + items = report.get("items") + item_count = len(items) if isinstance(items, list) else 0 + response = _build_run_lookup_response(record) + response.update( + { + "artifact": artifact, + "started_at": report.get("started_at"), + "completed_at": report.get("completed_at"), + "requested_limit": report.get("requested_limit"), + "pulled_count": report.get("pulled_count"), + "delivered_count": report.get("delivered_count"), + "marked_read_count": report.get("marked_read_count"), + "mark_read_requested": report.get("mark_read_requested"), + "debug_artifacts": report.get("debug_artifacts"), + "raw_output": _normalize_optional_repo_path(report.get("raw_output")), + "delivery_output": _normalize_optional_repo_path(report.get("delivery_output")), + "digest_brief_output": _normalize_optional_repo_path(report.get("digest_brief_output")), + "status_counts": report.get("status_counts"), + "keyword_index": _normalize_keyword_index(report.get("keyword_index")), + "item_count": item_count, + "report": report, + } + ) + return response + + def _resolve_run_record(run_id: str) -> dict[str, Any]: for run_dir in _iter_run_dirs(): record = _build_run_record(run_dir) @@ -235,6 +301,18 @@ def _build_status_response(record: dict[str, Any]) -> dict[str, Any]: } +def _build_run_lookup_response(record: dict[str, Any]) -> dict[str, Any]: + return { + "run_id": record["run_id"], + "run_dir": record["run_dir"].name, + "workflow": record["workflow"], + "run_type": record["run_type"], + "status": record["status"], + "output_dir": _normalize_repo_path(record["run_dir"]), + "state_source": record["state_source"], + } + + def _build_list_response(record: dict[str, Any]) -> dict[str, Any]: progress = _build_progress(record["stages"]) return { @@ -276,6 +354,85 @@ def _collect_artifacts(record: dict[str, Any]) -> list[dict[str, Any]]: return artifacts +def _resolve_artifact( + record: dict[str, Any], + *, + artifact_name: str, + relative_path: Path, + fallback_stage: str, + fallback_kind: str, + report_field: str | None = None, +) -> dict[str, Any]: + state = record.get("state") + if isinstance(state, RunState): + for artifact in state.artifacts: + if artifact.name != artifact_name: + continue + if _artifact_exists(artifact.path): + return _artifact_from_state_record(artifact) + + standard_path = record["run_dir"] / relative_path + if standard_path.exists(): + return _build_resolved_artifact( + name=artifact_name, + path=standard_path, + kind=fallback_kind, + stage=fallback_stage, + source="standard_path", + ) + + report = record.get("report") + if report_field and isinstance(report, dict): + report_path_value = report.get(report_field) + if isinstance(report_path_value, str) and report_path_value.strip(): + report_path = _resolve_repo_path(report_path_value) + if report_path.exists(): + return _build_resolved_artifact( + name=artifact_name, + path=report_path, + kind=fallback_kind, + stage=fallback_stage, + source="run_report_reference", + ) + + for path in sorted(record["run_dir"].rglob(relative_path.name)): + if path.is_file(): + return _build_resolved_artifact( + name=artifact_name, + path=path, + kind=fallback_kind, + stage=fallback_stage, + source="directory_scan", + ) + + available_artifacts = [artifact["name"] for artifact in _collect_artifacts(record)] + available_summary = ", ".join(available_artifacts) if available_artifacts else "none" + raise FileNotFoundError( + f"Artifact '{artifact_name}' not found for run_id: {record['run_id']} " + f"(status={record['status']}, run_dir={record['run_dir'].name}, available_artifacts={available_summary})" + ) + + +def _build_resolved_artifact( + *, + name: str, + path: Path, + kind: str, + stage: str, + source: str, +) -> dict[str, Any]: + return { + "name": name, + "kind": kind, + "stage": stage, + "path": _normalize_repo_path(path), + "exists": True, + "created_at": _iso_from_stat(path), + "metadata": {}, + "source": source, + } + + def _discover_artifacts(run_dir: Path) -> list[dict[str, Any]]: artifacts = [] for spec in DISCOVERED_ARTIFACTS: @@ -394,6 +551,22 @@ def _resolve_repo_path(path_value: str | Path) -> Path: return REPO_ROOT / path +def _normalize_optional_repo_path(path_value: Any) -> str | None: + if not isinstance(path_value, str) or not path_value.strip(): + return None + return _normalize_repo_path(path_value) + + +def _normalize_keyword_index(keyword_index: Any) -> dict[str, Any] | None: + if not isinstance(keyword_index, dict): + return None + + normalized = dict(keyword_index) + normalized["daily_output"] = _normalize_optional_repo_path(keyword_index.get("daily_output")) + normalized["stats_output"] = _normalize_optional_repo_path(keyword_index.get("stats_output")) + return normalized + + def _iso_from_stat(path: Path) -> str: return datetime.fromtimestamp(path.stat().st_mtime).astimezone().isoformat() diff --git a/src/summary_mcp/server.py b/src/summary_mcp/server.py index 6886c97..0d8df67 100644 --- a/src/summary_mcp/server.py +++ b/src/summary_mcp/server.py @@ -15,7 +15,9 @@ from summary_mcp.models.filtering import FilterContext, FilterInput from summary_mcp.models.item import Item from summary_mcp.models.llm_result import LlmSummaryResult from summary_mcp.models.summary_io import ExtractionInput +from summary_mcp.runtime import get_delivery_payload as load_delivery_payload from summary_mcp.runtime import get_run_status as load_run_status +from summary_mcp.runtime import get_run_report as load_run_report from summary_mcp.runtime import list_run_artifacts as load_run_artifacts from summary_mcp.runtime import list_runs as load_runs from summary_mcp.workflows import run_freshrss_pipeline @@ -144,6 +146,18 @@ def list_run_artifacts(run_id: str) -> dict: return load_run_artifacts(run_id=run_id) +@mcp.tool() +def get_delivery_payload(run_id: str) -> dict: + """Get the structured OpenClaw delivery payload for a workflow run.""" + return load_delivery_payload(run_id=run_id) + + +@mcp.tool() +def get_run_report(run_id: str) -> dict: + """Get the structured run report for a workflow run.""" + return load_run_report(run_id=run_id) + + @mcp.tool() def generate_article_summaries( *,