feat(oracle): export and apply evidence packets

README.md

@@ -144,6 +144,26 @@ python3 scripts/oracle_collection_queue.py \
   --format json
 ```
 
+如需让真人或 Antigravity AI 副手直接填写证据包，可一次性导出每个 case 的 draft JSON：
+
+```bash
+python3 scripts/oracle_collection_queue.py \
+  --oracle-file references/oracle/dasha_shadbala_oracle_cases.json \
+  --write-packet-dir references/oracle/artifacts/pending_packets \
+  --format json
+```
+
+填完某个 `external_*.json` 后，必须把 `status` 改为 `external_verified`，补齐 metadata、具体 `source_artifact` 文件路径以及所有 `target_placeholders`。再把该包合并回 oracle 文件：
+
+```bash
+python3 scripts/oracle_collection_queue.py \
+  --oracle-file references/oracle/dasha_shadbala_oracle_cases.json \
+  --apply-packet references/oracle/artifacts/pending_packets/external_template_steve_jobs_dasha_lahiri.json \
+  --format json
+```
+
+这一步只负责把人工填写的外部证据写回 `template_cases`；它不会自动认可证据，也不会允许生产调参。合并后仍必须重新生成 queue 并运行 validator。
+
 该 JSON 的 scope 是 `external_oracle_collection_queue`。当前队列有 5 个 `template_only` 任务、`ready_for_calibration: 0`、`production_tuning_allowed: false`，说明只能继续采集 JHora/PyJHora/VedAstro 等外部黑盒目标值；在模板字段未填充、状态未升为 `external_verified` 前，不能用这些样本做 Dasha/Shadbala 生产调参。
 
 Ashtakoot 外部合婚 oracle 使用同一个队列生成器，但独立样本文件是 `references/oracle/ashtakoot_oracle_cases.json`：

references/oracle/artifacts/README.md

@@ -32,3 +32,29 @@ inside external oracle evidence packets.
 Evidence packets remain `draft` until their target values are filled, their
 artifact is reviewed, and their status is explicitly promoted to
 `external_verified`.
+
+## Draft Packet Workflow
+
+Generate draft packets:
+
+```bash
+python3 scripts/oracle_collection_queue.py \
+  --oracle-file references/oracle/dasha_shadbala_oracle_cases.json \
+  --write-packet-dir references/oracle/artifacts/pending_packets \
+  --format json
+```
+
+Fill one `external_*.json` packet from JHora, PyJHora, VedAstro, or another
+documented external source. Then apply it back to the oracle file:
+
+```bash
+python3 scripts/oracle_collection_queue.py \
+  --oracle-file references/oracle/dasha_shadbala_oracle_cases.json \
+  --apply-packet references/oracle/artifacts/pending_packets/external_template_steve_jobs_dasha_lahiri.json \
+  --format json
+```
+
+Applying a packet does not make it trusted by itself. Rebuild the queue and run
+`scripts/oracle_evidence_validator.py`; only packets with complete metadata,
+filled targets, non-local artifacts, and `status: external_verified` can become
+`ready_for_calibration`.

scripts/oracle_collection_queue.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 import argparse
 import json
 import os
+import re
 from typing import Any
 
 
@@ -115,6 +116,12 @@ def _load_json(path: str) -> dict[str, Any]:
         return json.load(fh)
 
 
+def _write_json(path: str, value: dict[str, Any]) -> None:
+    with open(_resolve_path(path), "w", encoding="utf-8") as fh:
+        json.dump(value, fh, ensure_ascii=False, indent=2)
+        fh.write("\n")
+
+
 def _target_fields(value: Any, prefix: str = "target") -> list[str]:
     if prefix == "target" and isinstance(value, dict):
         return [f"{prefix}.{key}" for key in value]
@@ -178,6 +185,19 @@ def _evidence_packet(case: dict[str, Any], target_fields: list[str]) -> dict[str
     existing_placeholders = existing.get("target_placeholders", {})
     if isinstance(existing_placeholders, dict):
         target_placeholders.update(existing_placeholders)
+    default_metadata = {
+        "tool_name": "",
+        "tool_version_or_url": "",
+        "capture_date": "",
+        "source_artifact": "references/oracle/artifacts/",
+        "ayanamsa": case.get("settings", {}).get("ayanamsa", ""),
+        "node_mode": case.get("settings", {}).get("node_mode", ""),
+        "timezone": case.get("birth", {}).get("tz", ""),
+        "operator_note": "",
+    }
+    existing_metadata = existing.get("metadata", {})
+    if isinstance(existing_metadata, dict):
+        default_metadata.update(existing_metadata)
     return {
         "capture_id": existing.get("capture_id") or f"external_{case_id}",
         "status": existing.get("status", "draft"),
@@ -185,7 +205,7 @@ def _evidence_packet(case: dict[str, Any], target_fields: list[str]) -> dict[str
         "birth": case.get("birth", {}),
         "settings": case.get("settings", {}),
         "required_metadata_fields": REQUIRED_EVIDENCE_METADATA_FIELDS,
-        "metadata": existing.get("metadata", {}),
+        "metadata": default_metadata,
         "target_placeholders": target_placeholders,
         "integrity_checks": {
             "must_not_come_from_local_engine": True,
@@ -308,17 +328,93 @@ def render_markdown(queue: dict[str, Any]) -> str:
     return "\n".join(lines)
 
 
+def _safe_filename(value: str) -> str:
+    safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", value).strip("._")
+    return safe or "external_oracle_packet"
+
+
+def _set_target_value(target: dict[str, Any], field: str, value: Any) -> None:
+    parts = field.split(".")
+    if not parts or parts[0] != "target":
+        return
+    cursor = target
+    for part in parts[1:-1]:
+        child = cursor.get(part)
+        if not isinstance(child, dict):
+            child = {}
+            cursor[part] = child
+        cursor = child
+    cursor[parts[-1]] = value
+
+
+def apply_evidence_packet(oracle: dict[str, Any], packet: dict[str, Any]) -> bool:
+    case_id = packet.get("case_id")
+    if not case_id:
+        return False
+    for case in oracle.get("template_cases", []):
+        if case.get("id") != case_id and case.get("case_id") != case_id:
+            continue
+        target = case.setdefault("target", {})
+        placeholders = packet.get("target_placeholders", {})
+        if isinstance(placeholders, dict):
+            for field, value in placeholders.items():
+                _set_target_value(target, field, value)
+        case["status"] = packet.get("status", case.get("status", "draft"))
+        case["evidence_packet"] = {
+            "capture_id": packet.get("capture_id"),
+            "status": packet.get("status"),
+            "metadata": packet.get("metadata", {}),
+        }
+        return True
+    return False
+
+
+def write_evidence_packets(queue: dict[str, Any], output_dir: str) -> int:
+    os.makedirs(output_dir, exist_ok=True)
+    written = 0
+    for task in queue.get("tasks", []):
+        packet = task.get("evidence_packet", {})
+        capture_id = packet.get("capture_id") or f"external_{task.get('case_id', 'unknown')}"
+        packet_path = os.path.join(output_dir, f"{_safe_filename(str(capture_id))}.json")
+        with open(packet_path, "w", encoding="utf-8") as fh:
+            json.dump(packet, fh, ensure_ascii=False, indent=2)
+            fh.write("\n")
+        written += 1
+    return written
+
+
 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="Generate external oracle collection tasks")
     parser.add_argument("--oracle-file", required=True, help="Path to oracle fixture JSON")
     parser.add_argument("--format", choices=["json", "markdown"], default="json")
+    parser.add_argument(
+        "--write-packet-dir",
+        help="Optional directory where draft evidence_packet JSON files should be written.",
+    )
+    parser.add_argument(
+        "--apply-packet",
+        action="append",
+        default=[],
+        help="Filled evidence_packet JSON to merge back into the oracle file.",
+    )
     return parser.parse_args(argv)
 
 
 def main(argv: list[str] | None = None) -> int:
     args = parse_args(argv)
     oracle = _load_json(args.oracle_file)
+    applied_packets = 0
+    for packet_path in args.apply_packet:
+        packet = _load_json(packet_path)
+        if apply_evidence_packet(oracle, packet):
+            applied_packets += 1
+    if applied_packets:
+        _write_json(args.oracle_file, oracle)
     queue = build_queue(oracle)
+    if applied_packets:
+        queue["summary"]["applied_evidence_packets"] = applied_packets
+    if args.write_packet_dir:
+        queue["summary"]["written_evidence_packets"] = write_evidence_packets(queue, args.write_packet_dir)
     if args.format == "markdown":
         print(render_markdown(queue))
     else:

scripts/oracle_evidence_validator.py

@@ -66,6 +66,15 @@ def _is_blank(value: Any) -> bool:
     return value is None or value == "" or value == [] or value == {}
 
 
+def _is_missing_external_artifact(value: Any) -> bool:
+    if _is_blank(value):
+        return True
+    text = str(value).strip()
+    if text.endswith("/"):
+        return True
+    return text in {"references/oracle/artifacts", "references/oracle/artifacts/"}
+
+
 def _is_local_engine_artifact(packet: dict[str, Any]) -> bool:
     metadata = packet.get("metadata", {})
     haystack = " ".join(
@@ -85,7 +94,7 @@ def _validate_packet(task: dict[str, Any]) -> dict[str, Any]:
         if _is_blank(metadata.get(field)):
             problems.append(f"missing_metadata:{field}")
 
-    if _is_blank(metadata.get("source_artifact")):
+    if _is_missing_external_artifact(metadata.get("source_artifact")):
         problems.append("missing_external_artifact")
 
     target_placeholders = packet.get("target_placeholders", {})
@@ -108,7 +117,7 @@ def _validate_packet(task: dict[str, Any]) -> dict[str, Any]:
     if integrity.get("must_not_come_from_local_engine") and _is_local_engine_artifact(packet):
         problems.append("local_engine_artifact_rejected")
 
-    if integrity.get("requires_external_artifact") and _is_blank(metadata.get("source_artifact")):
+    if integrity.get("requires_external_artifact") and _is_missing_external_artifact(metadata.get("source_artifact")):
         if "missing_external_artifact" not in problems:
             problems.append("missing_external_artifact")
 

tests/test_oracle_collection_queue.py

@@ -102,6 +102,78 @@ def test_oracle_collection_queue_outputs_markdown_table() -> None:
     assert "capture_id" in markdown
 
 
+def test_oracle_collection_queue_can_write_draft_evidence_packets(tmp_path: Path) -> None:
+    completed = run_queue("--write-packet-dir", str(tmp_path))
+
+    assert completed.returncode == 0, completed.stderr or completed.stdout
+    report = json.loads(completed.stdout)
+    assert report["summary"]["written_evidence_packets"] == 5
+
+    packet_path = tmp_path / "external_template_steve_jobs_dasha_lahiri.json"
+    assert packet_path.exists()
+    packet = json.loads(packet_path.read_text(encoding="utf-8"))
+    assert packet["capture_id"] == "external_template_steve_jobs_dasha_lahiri"
+    assert packet["status"] == "draft"
+    assert packet["case_id"] == "template_steve_jobs_dasha_lahiri"
+    assert packet["metadata"]["source_artifact"] == "references/oracle/artifacts/"
+    assert packet["metadata"]["tool_name"] == ""
+    assert packet["target_placeholders"]["target.vimshottari_start_date"] is None
+    assert packet["integrity_checks"]["must_not_come_from_local_engine"] is True
+
+
+def test_oracle_collection_queue_can_apply_filled_evidence_packet(tmp_path: Path) -> None:
+    oracle = json.loads((ROOT / "references/oracle/dasha_shadbala_oracle_cases.json").read_text(encoding="utf-8"))
+    oracle_path = tmp_path / "oracle.json"
+    oracle_path.write_text(json.dumps(oracle, ensure_ascii=False), encoding="utf-8")
+
+    packet_dir = tmp_path / "packets"
+    generated = run_queue_for_file(oracle_path, "--write-packet-dir", str(packet_dir))
+    assert generated.returncode == 0, generated.stderr or generated.stdout
+
+    packet_path = packet_dir / "external_template_steve_jobs_dasha_lahiri.json"
+    packet = json.loads(packet_path.read_text(encoding="utf-8"))
+    packet["status"] = "external_verified"
+    packet["metadata"].update(
+        {
+            "tool_name": "JHora",
+            "tool_version_or_url": "manual-jhora-8.0",
+            "capture_date": "2026-06-26",
+            "source_artifact": "references/oracle/artifacts/steve_jobs_jhora_redacted.png",
+            "ayanamsa": "lahiri",
+            "node_mode": "true",
+            "timezone": "UTC-08:00",
+            "operator_note": "Typed from redacted external JHora screenshot.",
+        }
+    )
+    packet["target_placeholders"]["target.vimshottari_start_date"] = "1951-11-01"
+    packet["target_placeholders"]["target.shadbala_components"] = {
+        planet: {
+            "sthana": 1.0,
+            "dig": 2.0,
+            "kala": 3.0,
+            "chesta": 4.0,
+            "naisargika": 5.0,
+            "drik": 6.0,
+            "total_rupa": 21.0,
+        }
+        for planet in ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"]
+    }
+    packet_path.write_text(json.dumps(packet, ensure_ascii=False), encoding="utf-8")
+
+    applied = run_queue_for_file(oracle_path, "--apply-packet", str(packet_path))
+
+    assert applied.returncode == 0, applied.stderr or applied.stdout
+    report = json.loads(applied.stdout)
+    assert report["summary"]["applied_evidence_packets"] == 1
+
+    updated = json.loads(oracle_path.read_text(encoding="utf-8"))
+    case = next(row for row in updated["template_cases"] if row["id"] == "template_steve_jobs_dasha_lahiri")
+    assert case["status"] == "external_verified"
+    assert case["target"]["vimshottari_start_date"] == "1951-11-01"
+    assert case["target"]["shadbala_components"]["Sun"]["total_rupa"] == 21.0
+    assert case["evidence_packet"]["metadata"]["tool_name"] == "JHora"
+
+
 def test_first_jhora_evidence_packet_template_is_safe_and_fillable() -> None:
     template = ROOT / "references" / "oracle" / "evidence_packet_templates" / "jhora_steve_jobs_lahiri_first_packet.json"
     assert template.exists()