172 lines
7.4 KiB
Python
Executable File
172 lines
7.4 KiB
Python
Executable File
#!/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
|