181 lines
12 KiB
Python
181 lines
12 KiB
Python
"""Pure delivery projection of independently recorded read-only preflight."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
import re
|
|
|
|
from backend.browser.preflight_policy import compile_preflight_policy
|
|
|
|
|
|
_CLASSIFICATIONS = ("unique_dom_deep_reply", "needs_native_identity", "ambiguous", "unknown")
|
|
_NAV_STATUSES = {"entered_note", "revisited_note", "excluded_note_blocked", "note_limit_blocked", "unknown_click_blocked",
|
|
"action_not_allowed", "unexpected_note_rejected", "scope_history_untracked"}
|
|
_NAV_ACTIONS = {"navigate", "click", "input", "send_keys", "scroll", "go_back", "wait", "done", "ask_user",
|
|
"inspect_reply_relations", "advance_comments", "find_elements", "find_text", "search_page",
|
|
"screenshot", "observe", "other"}
|
|
|
|
|
|
def _valid_navigation(row: object) -> bool:
|
|
if not isinstance(row, dict) or set(row) != {"id", "version", "run_id", "goal_version", "status", "action", "note_id", "observed_at"}:
|
|
return False
|
|
if (type(row["version"]) is not int or row["version"] != 1
|
|
or not isinstance(row["id"], str) or not re.fullmatch(r"preflight_nav_[a-f0-9]{32}", row["id"])
|
|
or not isinstance(row["run_id"], str) or not re.fullmatch(r"[a-f0-9]{32}", row["run_id"])
|
|
or type(row["goal_version"]) is not int or not 1 <= row["goal_version"] <= 1_000_000_000
|
|
or not isinstance(row["status"], str) or row["status"] not in _NAV_STATUSES
|
|
or not isinstance(row["action"], str) or row["action"] not in _NAV_ACTIONS
|
|
or not isinstance(row["observed_at"], str) or len(row["observed_at"]) > 40):
|
|
return False
|
|
note_id = row["note_id"]
|
|
if note_id is not None and (not isinstance(note_id, str) or not re.fullmatch(r"[a-f0-9]{24}", note_id)):
|
|
return False
|
|
if row["status"] in {"entered_note", "revisited_note", "unexpected_note_rejected"} and note_id is None:
|
|
return False
|
|
try:
|
|
datetime.fromisoformat(row["observed_at"])
|
|
except ValueError:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _run_context(task: dict) -> tuple[int, str | None]:
|
|
inputs, artifacts = task.get("inputs") or {}, task.get("artifacts") or {}
|
|
version = inputs.get("_goal_version", 1)
|
|
if type(version) is not int or version < 1:
|
|
return 0, None
|
|
marker = artifacts.get("preflight_run")
|
|
if (not isinstance(marker, dict) or set(marker) != {"run_id", "goal_version"}
|
|
or marker.get("goal_version") != version or type(marker.get("goal_version")) is not int
|
|
or not isinstance(marker.get("run_id"), str) or not re.fullmatch(r"[a-f0-9]{32}", marker["run_id"])):
|
|
return version, None
|
|
return version, marker["run_id"]
|
|
|
|
|
|
def summarize_preflight(task: dict) -> dict | None:
|
|
"""Count only the current goal/run, without changing or upgrading history."""
|
|
from backend.browser.tools import _valid_preflight_telemetry
|
|
|
|
inputs, artifacts = task.get("inputs") or {}, task.get("artifacts") or {}
|
|
policy = compile_preflight_policy(inputs.get("goal", ""), inputs.get("max_notes"))
|
|
if not policy.active:
|
|
return None
|
|
goal_version, run_id = _run_context(task)
|
|
def current(row):
|
|
return isinstance(row, dict) and run_id is not None and row.get("run_id") == run_id and row.get("goal_version") == goal_version
|
|
raw_observations = artifacts.get("preflight_observations")
|
|
raw_navigation = artifacts.get("preflight_navigation")
|
|
observations = raw_observations if isinstance(raw_observations, list) else []
|
|
navigation = raw_navigation if isinstance(raw_navigation, list) else []
|
|
valid_observations = [row for row in observations if current(row) and _valid_preflight_telemetry(row)]
|
|
valid_navigation = [row for row in navigation if current(row) and _valid_navigation(row)]
|
|
prior_navigation = [row for row in navigation if _valid_navigation(row)]
|
|
malformed_current = (any(current(row) and not _valid_preflight_telemetry(row) for row in observations)
|
|
or any(current(row) and not _valid_navigation(row) for row in navigation))
|
|
# Wrapper receipts are append-only. A later unknown observation supersedes
|
|
# an earlier visible state; repeated checks never multiply sample counts.
|
|
latest = {}
|
|
for row in valid_observations:
|
|
latest[row["note_id"]] = row
|
|
visited = list(dict.fromkeys(row["note_id"] for row in valid_navigation if row["status"] in {"entered_note", "revisited_note"}))
|
|
all_observed_notes = set(visited) | set(latest)
|
|
order = [*visited, *(note_id for note_id in latest if note_id not in visited)]
|
|
counts = dict.fromkeys(_CLASSIFICATIONS, 0)
|
|
inspected, not_in_view, unknown, folded, folded_unknown = 0, 0, 0, 0, 0
|
|
checked = eligible = 0
|
|
notes = []
|
|
for ordinal, note_id in enumerate(order, 1):
|
|
row = latest.get(note_id)
|
|
confirmed = bool(row and row["status"] in {"checked", "no_visible_reply"} and row["discussion_state"] == "visible_comments")
|
|
if confirmed:
|
|
inspected += 1
|
|
checked += row["checked_count"]
|
|
eligible += row["eligible_count"]
|
|
for key in counts:
|
|
counts[key] += row["classification_counts"][key]
|
|
elif row and row["discussion_state"] == "comments_not_in_view":
|
|
not_in_view += 1
|
|
else:
|
|
unknown += 1
|
|
if row and row["has_unexpanded_replies"] is True:
|
|
folded += 1
|
|
elif not row or row["has_unexpanded_replies"] is None:
|
|
folded_unknown += 1
|
|
notes.append({"ordinal": ordinal, "inspected": confirmed,
|
|
"status": "inspected" if confirmed else "not_in_view" if row and row["discussion_state"] == "comments_not_in_view" else "unknown",
|
|
"checked_count": row["checked_count"] if confirmed else 0,
|
|
"eligible_count": row["eligible_count"] if confirmed else 0,
|
|
"has_unexpanded_replies": row["has_unexpanded_replies"] if row else None})
|
|
actual_violation = (any(row["status"] == "unexpected_note_rejected" for row in prior_navigation)
|
|
or bool(all_observed_notes & set(policy.excluded_note_ids))
|
|
or policy.max_notes is not None and len(all_observed_notes) > policy.max_notes)
|
|
truncated = any(row["history_truncated"] for row in valid_observations)
|
|
unknown_scope = (not valid_navigation or not visited or malformed_current or truncated
|
|
or bool(set(latest) - set(visited))
|
|
or any(row["status"] == "scope_history_untracked" for row in prior_navigation))
|
|
visited_known = bool(run_id and valid_navigation and visited and not malformed_current
|
|
and not any(row["status"] == "scope_history_untracked" for row in valid_navigation))
|
|
unknown_clicks = sum(row["status"] == "unknown_click_blocked" for row in valid_navigation)
|
|
scope_passed = False if actual_violation else None if unknown_scope else True
|
|
history = artifacts.get("browser_history") or {}
|
|
browser_success = artifacts.get("browser_verdict") == "success" and history.get("is_successful") is not False
|
|
completed = bool(browser_success and inspected > 0 and not not_in_view and not unknown and scope_passed is True)
|
|
if task.get("status") == "running":
|
|
headline = "正在预检可见回复关系"
|
|
elif completed:
|
|
headline = "预检已完成，发现可继续取证的样本" if eligible else "预检已完成，可见范围未发现合格样本"
|
|
elif inspected:
|
|
headline = "预检阶段结果已保留"
|
|
else:
|
|
headline = "尚未形成可核对的预检结果"
|
|
detail = (f"本段已记录实际访问{len(visited)}篇。" if visited_known else "本段实际访问数量未记录，访问范围无法完整核对。")
|
|
detail += (f"可确认预检{inspected}篇；检查当前可见回复{checked}条，其中{eligible}条具备继续取证的依据。"
|
|
if valid_observations else "本段尚无独立、可确认的预检记录。")
|
|
if not_in_view:
|
|
detail += f"{not_in_view}篇讨论尚未进入视口，不计入已预检。"
|
|
if unknown:
|
|
detail += f"{unknown}篇尚无可确认的当前预检结果。"
|
|
if folded:
|
|
detail += f"{folded}篇仍有折叠回复未展开。"
|
|
if folded_unknown:
|
|
detail += f"{folded_unknown}篇的折叠状态尚未确认。"
|
|
if unknown_clicks:
|
|
detail += f"已拒绝{unknown_clicks}个目标无法事前核对的入口，相关范围仍保留限制。"
|
|
if scope_passed is False:
|
|
detail += "已观察到超出本次访问范围的页面，本轮不能认定完成。"
|
|
elif scope_passed is None:
|
|
detail += "访问范围记录仍不完整，不能认定已遵守全部范围。"
|
|
detail += "零合格样本只对应本次可见范围，未展开或未加载部分仍未知；本轮未生成候选名单。"
|
|
return {"version": 1, "completed": completed, "inspected_notes": inspected, "visited_notes": len(visited),
|
|
"run_available": run_id is not None, "recorded": bool(valid_observations), "visited_known": visited_known,
|
|
"checked_count": checked, "eligible_count": eligible, "classification_counts": counts,
|
|
"not_in_view_notes": not_in_view, "unknown_notes": unknown, "folded_notes": folded,
|
|
"folded_unknown_notes": folded_unknown, "scope_passed": scope_passed,
|
|
"history_truncated": truncated, "max_notes": policy.max_notes,
|
|
"unknown_click_blocked_count": unknown_clicks,
|
|
"headline": headline, "detail": detail, "notes": notes}
|
|
|
|
|
|
def delivery_projection(task: dict) -> dict | None:
|
|
summary = summarize_preflight(task)
|
|
if summary is None:
|
|
return None
|
|
artifacts = task.get("artifacts") or {}
|
|
history = artifacts.get("browser_history") or {}
|
|
running = task.get("status") == "running"
|
|
status = "in_progress" if running else "fulfilled" if summary["completed"] else "partial" if summary["inspected_notes"] else "no_evidence"
|
|
return {"version": 1, "status": status, "headline": summary["headline"], "preflight": summary,
|
|
"runtime": {"status": task.get("status") or "unknown", "step": task.get("current_step"),
|
|
"stop_reason": artifacts.get("stop_reason"), "stop_detail": artifacts.get("stop_detail") or task.get("pause_reason") or "",
|
|
"raw_browser_success": history.get("is_successful"), "browser_verdict": artifacts.get("browser_verdict")},
|
|
"fulfillment": {"scope_met": summary["scope_passed"] is True, "requirements_met": summary["inspected_notes"] > 0 and not summary["not_in_view_notes"] and not summary["unknown_notes"],
|
|
"complete": summary["completed"], "assessment_current": True, "target": None, "current": 0, "remaining": None},
|
|
"artifacts": {"list_available": False, "file_required": False, "file_available": False,
|
|
"delivered": summary["inspected_notes"] > 0, "export_count": 0},
|
|
"counts": dict.fromkeys(("eligible", "known_identity", "supplemental", "unresolved", "signals", "verified_accounts"), 0),
|
|
"conditions": [
|
|
{"key": "preflight_visible_discussion", "label": "实际预检当前可见讨论", "required": True,
|
|
"met": summary["inspected_notes"] > 0 and not summary["not_in_view_notes"] and not summary["unknown_notes"], "current": summary["inspected_notes"], "target": None},
|
|
{"key": "preflight_visit_scope", "label": "实际访问符合本次范围", "required": True,
|
|
"met": summary["scope_passed"] is True, "current": summary["visited_notes"], "target": None}],
|
|
"people": []}
|