162 lines
11 KiB
Python
162 lines
11 KiB
Python
"""Program-owned operation facts and knowledge receipts; never trust model success."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from copy import deepcopy
|
|
|
|
PAGE_VARIANT = "xhs_visible_dom_v1"
|
|
|
|
|
|
def context_for(task: dict, *, stage: str | None = None) -> dict:
|
|
intent = (task.get("inputs") or {}).get("intent", "discover")
|
|
default_stage = "delivery" if intent == "export" else "outreach" if intent in {"follow", "message"} else "search_read"
|
|
return {"platform": "xiaohongshu", "stage": stage or default_stage,
|
|
"page_variant": PAGE_VARIANT, "authorized_capabilities": []}
|
|
|
|
|
|
def method_spec(capability: str, source_kind: str = "note") -> dict:
|
|
names = {"save_source": "保存独立来源", "expand_replies": "展开当前讨论回复", "inspect_profile": "核对来源主页号", "export": "生成当前名单文件"}
|
|
stage = "delivery" if capability == "export" else "search_read"
|
|
expected = {"save_source": "独立容器原话与层级核对通过并持久保存", "expand_replies": "当前父讨论中可见回复增加且归属可核对",
|
|
"inspect_profile": "来源作者入口匹配当前主页并取得带标签的可见小红书号", "export": "当前结果文件存在且清单与结果版本一致"}
|
|
return {"key": f"xhs_operation_{capability}_{source_kind}_v1", "version": 1,
|
|
"title": names[capability] + (f" · {source_kind}" if capability == "save_source" else ""),
|
|
"platform": "xiaohongshu", "stage": stage, "capability": capability, "source_kind": source_kind,
|
|
"page_variant": PAGE_VARIANT, "kind": "operation", "origin": "program_template",
|
|
"preconditions": ["当前页面实际可见且入口唯一", "按本次目标和授权执行"],
|
|
"parameters": ["current_observed_target"], "steps": ["检查当前实际入口", "执行限定操作", "核对变化并持久保存依据"],
|
|
"postcondition": expected[capability], "recovery": "无新增、定位失效或结果未知时停止重复操作，保留实际进度并重新观察。", "dependencies": []}
|
|
|
|
|
|
def operation_fact(capability: str, source_kind: str, object_key: str, pre: str, post: str,
|
|
*, outcome: str = "success", verifier: str, at: str, **details) -> dict:
|
|
# Only the program calls this after its tool checks. The model cannot supply
|
|
# these fields through save_learning/report_learning_outcome.
|
|
identity = hashlib.sha256(json.dumps([capability, source_kind, object_key, pre, post, outcome], ensure_ascii=False).encode()).hexdigest()
|
|
return {"id": identity, "capability": capability, "source_kind": source_kind,
|
|
"object_key": hashlib.sha256(object_key.encode()).hexdigest(), "outcome": outcome,
|
|
"evidence_refs": {"pre": pre, "post": post}, "verifier": verifier,
|
|
"observed_at": at, "page_variant": PAGE_VARIANT, "details": deepcopy(details)}
|
|
|
|
|
|
def _fact_trace(fact: dict) -> dict:
|
|
"""Use only the trace captured at execution; missing history stays missing."""
|
|
trace = (fact.get("details") or {}).get("trace")
|
|
if not isinstance(trace, dict):
|
|
return {}
|
|
result = {key: trace[key] for key in ("run_id", "action_id") if isinstance(trace.get(key), str) and trace[key]}
|
|
if type(trace.get("goal_version")) is int and trace["goal_version"] > 0:
|
|
result["goal_version"] = trace["goal_version"]
|
|
return result
|
|
|
|
|
|
def synchronize(task: dict, artifacts: dict) -> dict:
|
|
"""Register new executed facts only; old source snapshots are not backfilled."""
|
|
from backend.prospecting import knowledge
|
|
facts = artifacts.get("operation_evidence") or []
|
|
candidates = [row for row in artifacts.get("learning_candidates") or [] if isinstance(row, dict)
|
|
and row.get("evidence") and row.get("observed_at") and row.get("value")]
|
|
if not facts and not artifacts.get("knowledge_methods") and not candidates:
|
|
return artifacts
|
|
task_id = task["id"]
|
|
selected = {row["id"]: row for row in artifacts.get("knowledge_methods") or []}
|
|
registered = {}
|
|
events = []
|
|
fact_events = {}
|
|
valid_facts = []
|
|
for candidate in candidates:
|
|
digest = hashlib.sha256(json.dumps([candidate.get("key"), candidate["value"]], ensure_ascii=False).encode()).hexdigest()[:24]
|
|
spec = {"key": "suggested_" + digest, "version": 1, "title": str(candidate.get("key", "方法建议")),
|
|
"platform": "xiaohongshu", "stage": "search_read", "capability": "suggested_method", "source_kind": "any",
|
|
"page_variant": PAGE_VARIANT, "kind": "operation", "origin": "model",
|
|
"preconditions": ["当前实际页面必须重新观察"], "parameters": ["current_observed_target"],
|
|
"steps": [candidate["value"]], "postcondition": "需要匹配的程序验收器提供动作结果证明", "recovery": "没有独立动作验收时保留探索候选。", "dependencies": []}
|
|
try:
|
|
method = knowledge.register_method(spec, task_id)
|
|
registered[method["id"]] = method
|
|
except ValueError:
|
|
# Invalid task-specific text stays in the task candidate ledger,
|
|
# never in generic knowledge and never as a paid retry.
|
|
continue
|
|
for fact in facts:
|
|
if (not isinstance(fact, dict) or fact.get("capability") not in {"save_source", "expand_replies", "inspect_profile", "export"}
|
|
or not all(fact.get(key) for key in ("id", "object_key", "outcome", "verifier"))):
|
|
continue
|
|
spec = method_spec(fact["capability"], fact.get("source_kind", "note"))
|
|
method = knowledge.register_method(spec, task_id)
|
|
registered[method["id"]] = method
|
|
event = knowledge.record_event(method["id"], task_id, fact["object_key"], fact["outcome"],
|
|
evidence_refs=fact["evidence_refs"], context={**context_for(task, stage=spec["stage"]), "source_kind": spec["source_kind"]},
|
|
provenance="program", event_key="operation:" + fact["id"], verified=True,
|
|
verifier=fact["verifier"], **_fact_trace(fact))
|
|
events.append(event)
|
|
fact_events[fact["id"]] = event["id"]
|
|
if event.get("outcome") == "success":
|
|
valid_facts.append(fact)
|
|
# Merely injecting another method with the same capability is not
|
|
# proof it was adopted. It receives no automatic success credit.
|
|
# Agent reports remain reported/unknown, with no positive reliability count.
|
|
for report in artifacts.get("learning_outcomes") or []:
|
|
mid = str(report.get("entry_id", report.get("id", "")))
|
|
if mid not in selected:
|
|
continue
|
|
rid = hashlib.sha256(json.dumps(report, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
|
|
events.append(knowledge.record_event(mid, task_id, "reported:" + task_id,
|
|
"unused" if report.get("outcome") == "unused" else "unknown", provenance="model",
|
|
event_key="report:" + rid, reason="模型回执；未作为程序动作验收", evidence_refs={}))
|
|
# Stage success is scoped to this actual bounded branch, not a claim that
|
|
# the full search/profile/outreach chain or all comments were completed.
|
|
by_object = {}
|
|
for fact in valid_facts:
|
|
if fact.get("capability") == "save_source" and fact.get("outcome") == "success":
|
|
by_object.setdefault(fact["object_key"], {})[fact["source_kind"]] = fact
|
|
for object_key, kinds in by_object.items():
|
|
discussion = kinds.get("reply") or kinds.get("comment")
|
|
if "note" not in kinds or discussion is None:
|
|
continue
|
|
note = kinds["note"]
|
|
deps = [knowledge.register_method(method_spec("save_source", kind), task_id)["id"] for kind in ("note", discussion["source_kind"])]
|
|
spec = {"key": "xhs_stage_save_discussion_" + discussion["source_kind"] + "_v1", "version": 1,
|
|
"title": "保存一篇正文与独立讨论", "platform": "xiaohongshu", "stage": "search_read", "capability": "read_discussion",
|
|
"source_kind": discussion["source_kind"], "page_variant": PAGE_VARIANT, "kind": "pipeline", "origin": "program_template",
|
|
"preconditions": ["同一笔记已有实际可见正文和讨论"], "parameters": ["current_note"],
|
|
"steps": ["保存正文", "保存独立发言人的讨论"], "postcondition": "同笔记正文与一条独立讨论均保存通过",
|
|
"recovery": "保留已保存部分，未展开与未读取范围继续未知。", "dependencies": deps,
|
|
"branches": [discussion["source_kind"]], "terminal": "saved_note_and_attributed_discussion"}
|
|
method = knowledge.register_method(spec, task_id)
|
|
registered[method["id"]] = method
|
|
events.append(knowledge.record_event(method["id"], task_id, object_key, "success",
|
|
evidence_refs={"pre": note["evidence_refs"]["pre"], "post": discussion["evidence_refs"]["post"]},
|
|
context={**context_for(task, stage=spec["stage"]), "source_kind": discussion["source_kind"]}, provenance="program",
|
|
event_key="stage:" + note["id"] + ":" + discussion["id"], verified=True,
|
|
verifier="same_note_saved_sources", terminal_complete=True, executed_branch=discussion["source_kind"],
|
|
dependency_event_ids=[fact_events[note["id"]], fact_events[discussion["id"]]], **_fact_trace(discussion)))
|
|
acted = {row["method_id"] for row in events if row.get("provenance") == "program" and row.get("outcome") != "unused"}
|
|
reported = {str(row.get("entry_id", row.get("id", ""))) for row in artifacts.get("learning_outcomes") or []}
|
|
for mid in selected:
|
|
if mid not in reported and mid not in acted:
|
|
events.append(knowledge.record_event(mid, task_id, "task:" + task_id, "unused", provenance="program",
|
|
event_key="unused:" + task_id + ":" + mid, reason_code="not_observed_in_operation_evidence"))
|
|
return {**artifacts, "knowledge_summary": knowledge.task_summary(task_id),
|
|
"knowledge_method_ids": list(registered), "knowledge_event_ids": [row["id"] for row in events]}
|
|
|
|
|
|
def export_fact(task: dict, entry: dict) -> dict:
|
|
from backend.prospecting.workflows import export_path, result_version
|
|
path, metadata = export_path(task, entry["id"])
|
|
if not path.is_file() or metadata["result_version"] != result_version(task):
|
|
raise ValueError("文件与当前结果版本不一致，不能记录交付成功")
|
|
fact = operation_fact("export", "table", metadata["result_version"],
|
|
task["id"] + ":result:" + metadata["result_version"], task["id"] + ":export:" + entry["id"],
|
|
verifier="file_manifest_and_current_result", at=entry["generated_at"],
|
|
trace={"goal_version": metadata["goal_version"], "action_id": entry["id"]},
|
|
pre_snapshot={"result_version": metadata["result_version"], "goal_version": metadata["goal_version"]},
|
|
post_snapshot={"export_id": entry["id"], "exists": True, "current_version": True, "row_count": metadata["count"], "format": "csv"})
|
|
artifacts = dict(task.get("artifacts") or {})
|
|
facts = list(artifacts.get("operation_evidence") or [])
|
|
if not any(row.get("id") == fact["id"] for row in facts):
|
|
facts.append(fact)
|
|
artifacts["operation_evidence"] = facts
|
|
return artifacts
|