feat(web): authorize corrective smooth reader

my-tools/web_context.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""Bounded context helpers for the Blender/WBV2 task queue.
+
+The production context implementation lives in `tools/web/task-context-lib.mjs`.
+These helpers provide a small Python-compatible surface for local scripts and
+never load the full Blender source tree or historical evidence.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from pathlib import Path
+from typing import Any, Iterable
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DOCS = ROOT / "docs"
+TASKS = DOCS / "tasks"
+CONTROL_ROOT = ROOT / "tests/golden/WBV2"
+QUEUE = DOCS / "EXECUTION_QUEUE.md"
+INDEX = CONTROL_ROOT / "task-index.json"
+INVENTORY = CONTROL_ROOT / "dynamic-inventory.json"
+
+CARD_ID = re.compile(r"^#\s+(WBV2-P\d+-\d{3})", re.MULTILINE)
+CARD_STATUS = re.compile(r"^-\s*(?:status|状态):\s*`?([a-z_]+)`?", re.MULTILINE | re.IGNORECASE)
+PARENT = re.compile(r"^-\s*parent:\s*`?([^`\s]+)", re.MULTILINE | re.IGNORECASE)
+PATH_TOKEN = re.compile(r"(?:[A-Za-z0-9_.-]+/)+[A-Za-z0-9_.-]+(?:\.[A-Za-z0-9_.-]+)?")
+RAW_SUFFIXES = (".log", ".trace", ".out", ".err")
+
+
+def sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def load_cards() -> dict[str, tuple[Path, str, str]]:
+    cards: dict[str, tuple[Path, str, str]] = {}
+    for path in sorted(TASKS.glob("WBV2-P*.md")):
+        text = path.read_text(encoding="utf-8", errors="replace")
+        card = CARD_ID.search(text)
+        if not card:
+            continue
+        status_match = CARD_STATUS.search(text)
+        status = status_match.group(1).lower() if status_match else "unknown"
+        cards[card.group(1)] = (path, status, text)
+    return cards
+
+
+def predecessor_ids(text: str) -> list[str]:
+    parent = PARENT.search(text)
+    return [parent.group(1)] if parent and parent.group(1).startswith("WBV2-") else []
+
+
+def field_values(text: str, field_name: str) -> list[str]:
+    pattern = re.compile(rf"^-\s*{re.escape(field_name)}\s*[:：](.*)$", re.MULTILINE)
+    return [match.group(1).strip() for match in pattern.finditer(text)]
+
+
+def resolve_repo_path(candidate: str, origin: Path | None = None) -> Path | None:
+    candidate = candidate.strip().strip("`[]()")
+    if candidate.startswith(("http://", "https://")):
+        return None
+    candidates = []
+    if origin is not None:
+        candidates.append(origin.parent / candidate)
+    candidates.extend((ROOT / candidate, DOCS / candidate))
+    for path in candidates:
+        try:
+            path = path.resolve()
+            path.relative_to(ROOT.resolve())
+        except (OSError, ValueError):
+            continue
+        if path.is_file():
+            return path
+    return None
+
+
+def extract_referenced_files(texts: Iterable[tuple[str, Path | None]]) -> list[Path]:
+    found: dict[str, Path] = {}
+    for text, origin in texts:
+        for candidate in PATH_TOKEN.findall(text):
+            path = resolve_repo_path(candidate, origin)
+            if path:
+                found[str(path)] = path
+    return sorted(found.values(), key=lambda path: str(path))
+
+
+def classify_path(path: Path) -> tuple[str, str | None]:
+    rel = path.relative_to(ROOT).as_posix()
+    if path.name.endswith(RAW_SUFFIXES):
+        return "excluded", "raw log or trace"
+    if rel.startswith("blender-5.2.0/source/"):
+        return "excluded", "source tree is inspected by the task command"
+    if path.stat().st_size > 8 * 1024:
+        return "metadata_only", "evidence exceeds the context file cap"
+    return "read", None
+
+
+def _queue_files(task_id: str, card: tuple[Path, str, str]) -> list[tuple[str, Path]]:
+    files: list[tuple[str, Path]] = [("queue", QUEUE), ("task_card", card[0])]
+    manifest = CONTROL_ROOT.parent / task_id / "manifest.json"
+    if manifest.is_file():
+        files.append(("task_manifest", manifest))
+        value = json.loads(manifest.read_text(encoding="utf-8"))
+        parent = value.get("parentTask")
+        parent_manifest = CONTROL_ROOT.parent / str(parent) / "manifest.json" if parent else None
+        if parent_manifest and parent_manifest.is_file():
+            files.append(("parent_manifest", parent_manifest))
+    return files
+
+
+def context_manifest(task_id: str, *, include_evidence: bool = True, cards: dict[str, tuple[Path, str, str]] | None = None) -> dict[str, Any]:
+    cards = cards or load_cards()
+    if task_id not in cards:
+        raise KeyError(f"unknown task card: {task_id}")
+    card = cards[task_id]
+    entries: list[dict[str, Any]] = []
+    excluded: list[dict[str, str]] = []
+    for role, path in _queue_files(task_id, card):
+        kind, reason = classify_path(path)
+        if kind == "read":
+            entries.append({"role": role, "path": str(path.relative_to(ROOT)), "bytes": path.stat().st_size, "sha256": sha256(path), "mode": "read"})
+        else:
+            excluded.append({"path": str(path.relative_to(ROOT)), "reason": reason or kind})
+    if include_evidence:
+        for path in extract_referenced_files([(card[2], card[0])]):
+            if any(item["path"] == str(path.relative_to(ROOT)) for item in entries):
+                continue
+            kind, reason = classify_path(path)
+            if kind == "read":
+                entries.append({"role": "direct_evidence", "path": str(path.relative_to(ROOT)), "bytes": path.stat().st_size, "sha256": sha256(path), "mode": "read"})
+            else:
+                excluded.append({"path": str(path.relative_to(ROOT)), "reason": reason or kind})
+    total_bytes = sum(item["bytes"] for item in entries)
+    evidence = [item for item in entries if item["role"] == "direct_evidence"]
+    return {
+        "schema": "blender-web-task-context/v1",
+        "task_id": task_id,
+        "status": card[1],
+        "active_task": queue_pointer().get("task"),
+        "files": entries,
+        "excluded": excluded,
+        "totals": {"bytes": total_bytes, "estimated_tokens": total_bytes // 4, "evidence_bytes": sum(item["bytes"] for item in evidence), "evidence_files": len(evidence)},
+        "budget": {"total_bytes_limit": 16 * 1024, "evidence_bytes_limit": 12 * 1024, "evidence_file_limit": 6},
+        "read_order": ["queue", "task_card", "parent_manifest", "direct_evidence"],
+    }
+
+
+def queue_pointer() -> dict[str, str | None]:
+    text = QUEUE.read_text(encoding="utf-8")
+    task = re.search(r"\| 当前任务 \| `([^`]+)` \|", text)
+    parent = re.search(r"\| parent manifest \| `([^`]+)` \|", text, re.IGNORECASE)
+    return {"task": task.group(1) if task else None, "parentManifest": parent.group(1) if parent else None}
+
+
+def write_context_manifest(task_id: str, output: Path | None = None, manifest: dict[str, Any] | None = None) -> Path:
+    output = output or (CONTROL_ROOT.parent / task_id / "my-tools-context.json")
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(manifest or context_manifest(task_id), ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return output
+
+
+def write_context_summary(task_id: str, manifest: dict[str, Any] | None = None) -> Path:
+    manifest = manifest or context_manifest(task_id)
+    output = CONTROL_ROOT / "my-tools-context-summary.md"
+    lines = ["# Blender Web 最小续接上下文", "", f"- 当前任务：`{task_id}`（`{manifest['status']}`）", "- 只读取以下 mode=read 文件："]
+    lines.extend(f"  - `{item['path']}`" for item in manifest["files"] if item["mode"] == "read")
+    lines.append(f"- 清单 hash：`{sha256(write_context_manifest(task_id, manifest=manifest))}`")
+    output.write_text("\n".join(lines) + "\n", encoding="utf-8")
+    return output