ClaudeandClaude Opus 5.5
c9953c05f2
Product 2026-11-07: sync the astrologer's working branch (ea2a9d65..d09feab7). - 0266b6ad: year lord picks by the native five-fold Panchavargiya only; a PyJHora score (research switch) is attached per candidate as external_oracle_comparison (observation_only_not_selection_input) and never selects. panchavargiya_selection_policy on the result; candidate_list and the reader-report candidate table carry the five components and raw total. The legacy Sun/Mars/Saturn table relabel is not applicable: calc_sodhita_av was already removed here (BUG-1209). - 8c4bff8a + a5b75886: Sodhita no longer votes in finance adjudication; the bridge keeps raw 2/11 scores with prediction_scoring_allowed false. - ba8e36ec: finance delivery inventory (raw, no vote), wrapped producer status counts as blocked in machine-evidence sections, transit section, report gate evidence_claims bindings, finance support review readiness. - 5962368d: dispositor chain dispositor_sign is the dispositor's own sign. - 8813c851: BCP observed at the reading date (bcp_reference, PL9 row). - d09feab7: nakshatra_advanced.calculate_stellar_jeeva_sareera and the nadi-stellar-roles CLI; specialized-jyotish SKILL.md, usage packs, ontology and gap scanner mirrored. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DbBXoyn7Hj3LCW3BUs5bZ8
496 lines
28 KiB
Python
496 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
"""Compile public catalogues and local evidence into skill-family gap queues."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import json
|
|
import re
|
|
import urllib.request
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
REF = ROOT / "references"
|
|
|
|
CALCULATION_READINESS = {
|
|
"source_bounded_wired": 32,
|
|
"native_candidate": 30,
|
|
"formula_partial": 24,
|
|
"unsafe_legacy_stub_external_replay_available": 12,
|
|
"external_replay_available": 12,
|
|
"report_surface_observed": 14,
|
|
"text_hit_unclassified": 8,
|
|
"catalogued": 5,
|
|
"registered_placeholder": 4,
|
|
"absent": 0,
|
|
}
|
|
FAMILY_VALUE = {"dasha": 18, "kp": 17, "pravesha": 16, "sphuta": 12, "chakra": 11, "panchanga": 14}
|
|
NON_IMPLEMENTATION_SCRIPTS = {
|
|
"scripts/audit_local_knowledge_copies.py",
|
|
"scripts/book_rule_lineage.py",
|
|
"scripts/technique_gap_scanner.py",
|
|
"scripts/technique_source_adapters.py",
|
|
}
|
|
GENERIC_BOOK_ALIASES = {"dasha", "dasa", "mahadasha", "antardasha", "pratyantardasha"}
|
|
|
|
# Positive completion states: new/blocked statuses must never silently disappear.
|
|
COMPLETE_DIMENSION_VALUES = {
|
|
"identity_status": {"canonicalized", "resolved"},
|
|
"doctrine_status": {"usage_pack_present"},
|
|
"variant_status": {"profile_inventory_present"},
|
|
"calculation_status": {"source_bounded_wired"},
|
|
"routing_status": {"family_assigned", "skill_linked"},
|
|
"report_status": {"source_bounded_wired"},
|
|
}
|
|
|
|
|
|
def missing_dimensions(dimensions: dict[str, str]) -> list[str]:
|
|
return [key for key, accepted in COMPLETE_DIMENSION_VALUES.items()
|
|
if dimensions.get(key) not in accepted]
|
|
|
|
|
|
def entrypoint_exists(entrypoint: str) -> bool:
|
|
"""Static function presence only; never imports or claims successful execution."""
|
|
module, _, name = entrypoint.rpartition('.')
|
|
path = ROOT / (module.replace('.', '/') + '.py')
|
|
try:
|
|
tree = ast.parse(path.read_text(encoding='utf-8'))
|
|
except (OSError, SyntaxError):
|
|
return False
|
|
return any(isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name
|
|
for node in tree.body)
|
|
|
|
|
|
def scoped_work(technique_id, override, review):
|
|
scopes, tasks = [], []
|
|
for contract_id, contract in override.items():
|
|
if not contract_id.endswith('_contract') or not isinstance(contract, dict):
|
|
continue
|
|
entries = [(key, value) for key, value in contract.items()
|
|
if key.endswith('entrypoint') and isinstance(value, str)]
|
|
entries += [('entrypoints', value) for value in contract.get('entrypoints', [])]
|
|
for role, entrypoint in entries:
|
|
exists = entrypoint_exists(entrypoint)
|
|
scope_id = f'{technique_id}:{contract_id}:{entrypoint}'
|
|
scopes.append(dict(scope_id=scope_id, contract_id=contract_id,
|
|
entrypoint=entrypoint, role=role, entrypoint_present=exists,
|
|
execution_status='not_run_by_scanner',
|
|
implementation_kind='external_observation' if 'external' in contract_id else 'native_or_composed',
|
|
contract=contract, usage=review['usage']))
|
|
if not exists:
|
|
tasks.append(dict(task_id=scope_id+':restore', technique_id=technique_id,
|
|
lane='development', status='ready', target=entrypoint,
|
|
acceptance='Restore declared callable entrypoint and run its calculation/report regression.'))
|
|
for name, status, target, acceptance in review['development']:
|
|
tasks.append(dict(task_id=f'{technique_id}:{name}', technique_id=technique_id,
|
|
lane='development', status=status, target=target, acceptance=acceptance))
|
|
if not scopes:
|
|
tasks.insert(0, dict(task_id=f'{technique_id}:restore_declared_scope',
|
|
technique_id=technique_id, lane='development', status='ready',
|
|
target='Restore or declare at least one callable capability scope.',
|
|
acceptance='Declare a real entrypoint with input/output contract and add a focused calculation/report regression.'))
|
|
tasks.append(dict(task_id=technique_id+':source_research', technique_id=technique_id,
|
|
lane='research', status='open', target=review['research'],
|
|
acceptance='Record exact source/profile and independent example, or a specific unresolved discrepancy; no capability promotion by label.'))
|
|
tasks.append(dict(task_id=technique_id+':customer_review', technique_id=technique_id,
|
|
lane='customer_admission', status='not_authorized_in_this_workflow',
|
|
target='Separate report-router review for the selected profile; default export unchanged.',
|
|
acceptance='Explicit profile-specific router approval and customer-chain regression in the owning workflow.'))
|
|
return scopes, tasks
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def normalized(value: str) -> str:
|
|
return re.sub(r"[^a-z0-9]+", "", value.lower())
|
|
|
|
|
|
def _priority(row: dict[str, Any]) -> tuple[int, str, str, list[str]]:
|
|
dimensions = row["dimensions"]
|
|
calculation = dimensions["calculation_status"]
|
|
reasons = []
|
|
score = 20 + FAMILY_VALUE.get(row["family"], 8) + CALCULATION_READINESS.get(calculation, 0)
|
|
if row.get("documented_methods"):
|
|
score += 10
|
|
reasons.append("public method surface documented")
|
|
if row.get("book_rule_ids") or row.get("book_topic_ids"):
|
|
score += 8
|
|
reasons.append("local book linkage available")
|
|
if row.get("local_implementation_hits") or row.get("declared_local_paths"):
|
|
score += 8
|
|
reasons.append("local implementation evidence available")
|
|
if dimensions["identity_status"] == "needs_canonicalization":
|
|
score -= 18
|
|
reasons.append("identity must be canonicalized first")
|
|
if calculation == "source_bounded_wired":
|
|
phase, next_work = "P0", "complete_variant_and_report_contract"
|
|
elif calculation in {"native_candidate", "formula_partial"}:
|
|
phase, next_work = "P0", "verify_or_wire_existing_calculation"
|
|
elif calculation in {"unsafe_legacy_stub_external_replay_available", "external_replay_available"}:
|
|
phase, next_work = "P1", "complete_native_doctrine_or_keep_external_only"
|
|
elif calculation == "report_surface_observed":
|
|
phase, next_work = "P1", "recover_or_build_producer_from_report_surface"
|
|
elif row.get("documented_methods") or row.get("book_rule_ids") or row.get("book_topic_ids"):
|
|
phase, next_work = "P1", "complete_doctrine_and_variant_pack"
|
|
else:
|
|
phase, next_work = "P2", "research_identity_doctrine_and_formula"
|
|
reasons.insert(0, f"calculation readiness: {calculation}")
|
|
return max(0, min(score, 100)), phase, next_work, reasons
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _local_corpus() -> dict[str, tuple[str, str]]:
|
|
paths = [
|
|
REF / "technique_registry.json",
|
|
REF / "book_rule_technique_lineage.json",
|
|
REF / "book_rule_packets.json",
|
|
REF / "internet_technique_gap_registry.json",
|
|
]
|
|
raw = {str(path.relative_to(ROOT)): path.read_text(encoding="utf-8", errors="ignore") for path in paths if path.exists()}
|
|
for path in (ROOT / "scripts").glob("*.py"):
|
|
raw[str(path.relative_to(ROOT))] = path.read_text(encoding="utf-8", errors="ignore")
|
|
return {path: (text.lower(), normalized(text)) for path, text in raw.items()}
|
|
|
|
|
|
def _implementation_paths(path_hits: list[str]) -> list[str]:
|
|
return [path for path in path_hits if path.startswith("scripts/") and path not in NON_IMPLEMENTATION_SCRIPTS]
|
|
|
|
|
|
def _book_topic_matches(topic: dict[str, Any], compact_terms: set[str]) -> bool:
|
|
for alias in topic.get("aliases", []):
|
|
compact_alias = normalized(str(alias))
|
|
if not compact_alias or compact_alias in GENERIC_BOOK_ALIASES:
|
|
continue
|
|
if any(compact_alias == term or compact_alias in term for term in compact_terms):
|
|
return True
|
|
return False
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def scan() -> dict[str, Any]:
|
|
catalog = load_json(REF / "external_technique_catalog.json")
|
|
ontology = load_json(REF / "technique_ontology.json")
|
|
templates = load_json(REF / "technique_family_templates.json")["families"]
|
|
variant_registry = load_json(REF / "technique_variant_profiles.json").get("techniques", {})
|
|
gap_rows = {row["id"]: row for row in load_json(REF / "internet_technique_gap_registry.json")["techniques"]}
|
|
aliases = ontology.get("aliases", {})
|
|
overrides = ontology.get("overrides", {})
|
|
reviews = ontology.get('work_scope_reviews', {})
|
|
# Keep ``development`` as a compatibility aggregate. Active and blocked
|
|
# development are separate queues so source uncertainty does not hide
|
|
# work that can be safely implemented with an adapter or explicit profile.
|
|
work_queues = {lane: [] for lane in (
|
|
'development', 'active_development', 'blocked_development',
|
|
'research', 'customer_admission')}
|
|
lineage_rows = load_json(REF / "book_rule_technique_lineage.json").get("rows", [])
|
|
skill_families = load_json(REF / "skill_family_router.json").get("families", {})
|
|
discovered_path = REF / "external_technique_discovery_snapshot.json"
|
|
discovered = load_json(discovered_path).get("candidates", []) if discovered_path.exists() else []
|
|
discovery_by_id: dict[str, dict[str, Any]] = {}
|
|
for candidate in discovered:
|
|
technique_id = candidate.get("canonical_id") or candidate["candidate_id"]
|
|
existing = discovery_by_id.setdefault(technique_id, {})
|
|
for key in ("documented_module", "documented_methods", "documented_parameters", "applicability_hints"):
|
|
value = candidate.get(key)
|
|
if value and not existing.get(key):
|
|
existing[key] = value
|
|
book_topics_path = REF / "local_book_technique_candidates.json"
|
|
book_topics = load_json(book_topics_path).get("topics", []) if book_topics_path.exists() else []
|
|
external_rows = list(catalog["techniques"])
|
|
known_ids = {row["canonical_id"] for row in external_rows}
|
|
for candidate in discovered:
|
|
technique_id = candidate.get("canonical_id") or candidate["candidate_id"]
|
|
if technique_id in known_ids:
|
|
continue
|
|
external_rows.append({
|
|
"canonical_id": technique_id,
|
|
"name": candidate["name"],
|
|
"family": candidate["family"],
|
|
"skill_family": candidate["skill_family"],
|
|
"sources": [candidate["source_id"]],
|
|
"identity_status": candidate["resolution_status"],
|
|
})
|
|
known_ids.add(technique_id)
|
|
texts = _local_corpus()
|
|
rows = []
|
|
for external in external_rows:
|
|
technique_id = external["canonical_id"]
|
|
source_metadata = discovery_by_id.get(technique_id, {})
|
|
terms = [technique_id.replace("_", " "), external["name"], *aliases.get(technique_id, [])]
|
|
normalized_terms = [(term.lower(), normalized(term)) for term in terms]
|
|
path_hits = sorted(path for path, (lower_text, normalized_text) in texts.items()
|
|
if any(lower in lower_text or compact in normalized_text for lower, compact in normalized_terms))
|
|
implementation_hits = _implementation_paths(path_hits)
|
|
override = overrides.get(technique_id, {})
|
|
declared_variant_ids = override.get("variant_profile_ids", [])
|
|
available_variant_profiles = {
|
|
profile["profile_id"]: profile
|
|
for profile in variant_registry.get(technique_id, {}).get("profiles", [])
|
|
}
|
|
missing_variant_ids = [profile_id for profile_id in declared_variant_ids if profile_id not in available_variant_profiles]
|
|
if missing_variant_ids:
|
|
raise ValueError(f"{technique_id}: missing variant profiles: {', '.join(missing_variant_ids)}")
|
|
gap = gap_rows.get(technique_id, {})
|
|
calculation = override.get("calculation_status") or gap.get("status") or ("text_hit_unclassified" if implementation_hits else "absent")
|
|
family_template = templates.get(external["family"], {})
|
|
dimensions = {
|
|
"identity_status": external.get("identity_status", "canonicalized"),
|
|
"discovery_status": "catalogued",
|
|
"doctrine_status": override.get("doctrine_status", "usage_required"),
|
|
"variant_status": override.get("variant_status", "variant_inventory_required"),
|
|
"calculation_status": calculation,
|
|
"routing_status": override.get("routing_status", "family_assigned"),
|
|
"report_status": override.get("report_status", "not_verified"),
|
|
"source_status": "public_catalogued",
|
|
}
|
|
missing = missing_dimensions(dimensions)
|
|
compact_terms = {normalized(term) for term in terms}
|
|
book_rules = [row["rule_id"] for row in lineage_rows
|
|
if any(normalized(str(technique)) in compact_terms or any(term in normalized(str(technique)) for term in compact_terms)
|
|
for technique in row.get("techniques", []))]
|
|
book_topic_ids = [topic["topic_id"] for topic in book_topics if _book_topic_matches(topic, compact_terms)]
|
|
row = {
|
|
**external,
|
|
**source_metadata,
|
|
"aliases": aliases.get(technique_id, []),
|
|
"dimensions": dimensions,
|
|
"local_text_hits": path_hits,
|
|
"local_implementation_hits": implementation_hits,
|
|
"book_rule_ids": book_rules,
|
|
"book_topic_ids": book_topic_ids,
|
|
"declared_local_paths": override.get("local_paths", [gap.get("local_path")] if gap.get("local_path") else []),
|
|
"producer_contract": override.get("producer_contract", {}),
|
|
"external_replay_contract": override.get("external_replay_contract", {}),
|
|
"variant_profiles": [available_variant_profiles[profile_id] for profile_id in declared_variant_ids],
|
|
"report_paths": override.get("report_paths", []),
|
|
"next_required": override.get("next_required", []),
|
|
"required_inputs": override.get("required_inputs", gap.get("inputs", family_template.get("required_inputs", []))),
|
|
"use_for": override.get("use_for", gap.get("use_for", [])),
|
|
"never_use_as": override.get("never_use_as", gap.get("not_for", family_template.get("never_use_as", []))),
|
|
"required_rule_fields": override.get("required_rule_fields", family_template.get("required_rule_fields", [])),
|
|
"missing_dimensions": missing,
|
|
}
|
|
if technique_id in reviews:
|
|
scopes, tasks = scoped_work(technique_id, override, reviews[technique_id])
|
|
row['capability_scopes'] = scopes
|
|
row['work_scope_reviewed'] = True
|
|
row['work_items'] = tasks
|
|
for task in tasks:
|
|
work_queues[task['lane']].append(task)
|
|
if task['lane'] == 'development':
|
|
lane = ('blocked_development'
|
|
if task['status'] == 'blocked_on_source'
|
|
else 'active_development')
|
|
work_queues[lane].append(task)
|
|
score, phase, next_work, reasons = _priority(row)
|
|
row.update({
|
|
"implementation_priority_score": score,
|
|
"implementation_phase": phase,
|
|
"next_work_type": next_work,
|
|
"priority_reasons": reasons,
|
|
"priority_semantics": "engineering implementation order only; not astrological validity or authority",
|
|
})
|
|
rows.append(row)
|
|
queues: dict[str, list[str]] = {family: [] for family in skill_families}
|
|
for row in rows:
|
|
if row["missing_dimensions"]:
|
|
queues.setdefault(row["skill_family"], []).append(row["canonical_id"])
|
|
by_id = {row["canonical_id"]: row for row in rows}
|
|
for items in queues.values():
|
|
items.sort(key=lambda technique_id: (-by_id[technique_id]["implementation_priority_score"], technique_id))
|
|
queue_phase_counts = {
|
|
phase: sum(by_id[item]["implementation_phase"] == phase for items in queues.values() for item in items)
|
|
for phase in ("P0", "P1", "P2")
|
|
}
|
|
return {
|
|
"schema": "jyotish.technique_gap_scan.v1",
|
|
"work_queues": work_queues,
|
|
"work_queue_coverage": {'reviewed_techniques': sorted(reviews),
|
|
'boundary': 'Reviewed families only; other legacy requirements remain visible, not globally triaged.'},
|
|
"summary": {
|
|
"queue_count_semantics": "legacy_family_dimension_gaps_not_development_tasks",
|
|
"work_queue_counts": {lane: len(work_queues[lane]) for lane in
|
|
('active_development', 'blocked_development',
|
|
'research', 'customer_admission')},
|
|
"active_development_count": len(work_queues['active_development']),
|
|
"ready_development_count": sum(t['status'] == 'ready' for t in work_queues['active_development']),
|
|
"blocked_development_count": len(work_queues['blocked_development']),
|
|
"external_catalogue_count": len(rows),
|
|
"curated_catalogue_count": len(catalog["techniques"]),
|
|
"live_discovery_count": len(discovered),
|
|
"unresolved_identity_count": sum(row["dimensions"]["identity_status"] == "needs_canonicalization" for row in rows),
|
|
"local_book_topic_count": len(book_topics),
|
|
"queue_count": sum(len(items) for items in queues.values()),
|
|
"queue_phase_counts": queue_phase_counts,
|
|
"skill_family_count": len(queues),
|
|
"variant_profile_count": sum(len(item.get("profiles", [])) for item in variant_registry.values()),
|
|
"global_closure_claimed": False,
|
|
},
|
|
"techniques": rows,
|
|
"queues": queues,
|
|
"local_book_topics": book_topics,
|
|
}
|
|
|
|
|
|
def check_web() -> dict[str, Any]:
|
|
sources = load_json(REF / "external_technique_catalog.json")["sources"]
|
|
result = {}
|
|
for source_id, url in sources.items():
|
|
try:
|
|
request = urllib.request.Request(url, headers={"User-Agent": "jyotish-technique-gap-scanner/1.0"})
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
result[source_id] = {"url": url, "status": "reachable", "http_status": response.status}
|
|
except Exception as exc: # network state is evidence, never a technique absence claim
|
|
result[source_id] = {"url": url, "status": "blocked", "reason": type(exc).__name__}
|
|
return result
|
|
|
|
|
|
def render_queue(skill_family: str, rows: list[dict[str, Any]]) -> str:
|
|
lines = ["# Generated technique gap queue", "", "Generated from `scripts/technique_gap_scanner.py`; edit source registries, then regenerate.", "", "Legacy dimension gaps are not development task counts. Reviewed work is separated below; this scanner only inspects declarations.", "", *render_scoped_work(rows), "## Unreviewed legacy dimension gaps", "", "| Technique | Phase | Score | Family | Calculation | Next work | Missing dimensions |", "|---|---:|---:|---|---|---|---|"]
|
|
for row in rows:
|
|
if row.get('work_scope_reviewed'):
|
|
continue
|
|
lines.append(f"| {row['name']} | {row['implementation_phase']} | {row['implementation_priority_score']} | {row['family']} | {row['dimensions']['calculation_status']} | {row['next_work_type']} | {', '.join(row['missing_dimensions'])} |")
|
|
if not rows:
|
|
lines.extend(["", "No candidate from the current external catalogue. This does not prove the family has no global gaps."])
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def render_scoped_work(rows):
|
|
reviewed = [row for row in rows if row.get('work_scope_reviewed')]
|
|
if not reviewed:
|
|
return []
|
|
lines = ['## Capability-scoped work', '',
|
|
'Static entrypoint presence is not a runtime claim: not_run_by_scanner. Original maturity/profile limits remain in the referenced contracts and usage instructions. Counts cover reviewed families only.', '']
|
|
for row in reviewed:
|
|
lines += [f"### {row['name']}", '', '| Contract / profile scope | Entry | Present | Usage |', '|---|---|---|---|']
|
|
for scope in row['capability_scopes']:
|
|
lines.append(f"| {scope['contract_id']} | `{scope['entrypoint']}` | {scope['entrypoint_present']} | `{scope['usage']}` |")
|
|
lines.append('')
|
|
for lane in ('active_development', 'blocked_development', 'research', 'customer_admission'):
|
|
lines += [f'### {lane}', '', '| Task | Status | Target | Acceptance |', '|---|---|---|---|']
|
|
for row in reviewed:
|
|
for task in row['work_items']:
|
|
task_lane = task['lane']
|
|
if task_lane == 'development':
|
|
task_lane = ('blocked_development'
|
|
if task['status'] == 'blocked_on_source'
|
|
else 'active_development')
|
|
if task_lane == lane:
|
|
lines.append(f"| {task['task_id']} | {task['status']} | {task['target']} | {task['acceptance']} |")
|
|
lines.append('')
|
|
return lines
|
|
|
|
|
|
def render_usage_pack(skill_family: str, rows: list[dict[str, Any]]) -> str:
|
|
lines = ["# Generated technique usage pack", "", "Generated from ontology, family templates and public interface documentation. It is an intake contract, not invented doctrine. Public method metadata does not prove local executability or doctrinal correctness.", ""]
|
|
if not rows:
|
|
lines.append("No usage package generated from the current external catalogue.")
|
|
for row in rows:
|
|
replay = row.get("external_replay_contract", {})
|
|
lines.extend([
|
|
f"### {row['name']}", "",
|
|
f"- Canonical ID: `{row['canonical_id']}`",
|
|
f"- Legacy family priority (not scoped development count): {row['implementation_phase']} / {row['implementation_priority_score']}",
|
|
f"- Next work: {'See capability-scoped work below' if row.get('work_scope_reviewed') else row['next_work_type']}",
|
|
f"- Remaining requirements: {'; '.join(row.get('next_required', [])) or 'See missing dimensions and source-specific usage notes'}",
|
|
f"- Native producer: `{row.get('producer_contract', {}).get('entrypoint') or 'not established'}`",
|
|
f"- Producer maturity: `{row.get('producer_contract', {}).get('maturity') or 'not established'}`",
|
|
f"- Producer verification scope: {row.get('producer_contract', {}).get('verification_scope') or 'not established'}",
|
|
f"- Report paths: {', '.join(f'`{item}`' for item in row.get('report_paths', [])) or 'not wired or not verified'}",
|
|
f"- Variant profiles: {', '.join('`{}` ({})'.format(item['profile_id'], item['status']) for item in row.get('variant_profiles', [])) or 'inventory required'}",
|
|
f"- Required inputs: {', '.join(row['required_inputs']) or 'family template pending'}",
|
|
f"- Use for: {', '.join(row['use_for']) or 'doctrine pack required before interpretation'}",
|
|
f"- Never use as: {', '.join(row['never_use_as'])}",
|
|
f"- Required rule fields: {', '.join(row['required_rule_fields'])}",
|
|
f"- Local evidence hits: {', '.join(row['local_text_hits']) or 'none'}",
|
|
f"- Local implementation-candidate hits: {', '.join(row['local_implementation_hits']) or 'none'}",
|
|
f"- Book rule links: {', '.join(row['book_rule_ids']) or 'not mapped'}",
|
|
f"- Book topic links: {', '.join(row['book_topic_ids']) or 'not mapped'}",
|
|
f"- Documented external module: `{row.get('documented_module') or 'not documented'}`",
|
|
f"- Documented external methods: {', '.join(f'`{item}`' for item in row.get('documented_methods', [])) or 'not documented'}",
|
|
f"- Documented parameters: {', '.join(f'`{item}`' for item in row.get('documented_parameters', [])) or 'not documented'}",
|
|
f"- Applicability hints from source documentation: {'; '.join(row.get('applicability_hints', [])) or 'none extracted; doctrine review required'}",
|
|
])
|
|
if replay:
|
|
lines.extend([
|
|
f"- External replay: `{replay.get('entrypoint')}`",
|
|
f"- Separate external report entrypoint: `{replay.get('report_entrypoint') or 'not separately declared; inspect replay contract'}`",
|
|
f"- External replay status: `{replay.get('status')}`",
|
|
f"- External replay boundary: {replay.get('claim_boundary')}",
|
|
])
|
|
if row.get('work_scope_reviewed'):
|
|
lines.extend(render_scoped_work([row]))
|
|
lines.append("")
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def render_roadmap(report: dict[str, Any]) -> str:
|
|
rows = sorted((row for row in report["techniques"] if row["missing_dimensions"] and not row.get('work_scope_reviewed')), key=lambda row: (-row["implementation_priority_score"], row["canonical_id"]))
|
|
lines = [
|
|
"# Technique integration roadmap", "",
|
|
"Generated from the current ontology, live public catalogue metadata and local evidence. Scores order engineering work only; they do not rank astrological validity, predictive power or textual authority.", "",
|
|
]
|
|
lines += render_scoped_work(report['techniques'])
|
|
lines += ['## Legacy requirements not yet scope-reviewed', '',
|
|
'The compatibility queue_count counts family dimension gaps, not development tasks. Below are unreviewed families; absence does not establish global completion.', '']
|
|
for phase, meaning in (
|
|
("P0", "Repair, verify or wire calculations already present locally."),
|
|
("P1", "Complete doctrine, variant and usage packs before implementing."),
|
|
("P2", "Research identity, doctrine and formula before implementation."),
|
|
):
|
|
lines.extend([f"## {phase}", "", meaning, "", "| Technique | Skill | Score | Next work |", "|---|---|---:|---|"])
|
|
for row in rows:
|
|
if row["implementation_phase"] == phase:
|
|
lines.append(f"| {row['name']} | {row['skill_family']} | {row['implementation_priority_score']} | {row['next_work_type']} |")
|
|
lines.append("")
|
|
# Wiring a finite profile must not erase its still-open source/case work.
|
|
followups = [row for row in report["techniques"]
|
|
if not row["missing_dimensions"] and row.get("next_required")]
|
|
lines += ["## Wired scopes with remaining work", "",
|
|
"These are outside the implementation queue, not globally complete. Original source, variant and verification follow-ups remain actionable.", ""]
|
|
for row in sorted(followups, key=lambda item: item['canonical_id']):
|
|
lines += [f"### {row['name']}", "", *[f"- {item}" for item in row['next_required']], ""]
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def write_outputs(report: dict[str, Any], output: Path) -> None:
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
by_id = {row["canonical_id"]: row for row in report["techniques"]}
|
|
for family, ids in report["queues"].items():
|
|
target = ROOT / "skills" / family / "references" / "generated-gap-queue.md"
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(render_queue(family, [by_id[item] for item in ids]), encoding="utf-8")
|
|
usage_target = ROOT / "skills" / family / "references" / "generated-usage-pack.md"
|
|
family_rows = sorted(
|
|
(row for row in report["techniques"] if row["skill_family"] == family),
|
|
key=lambda row: (-row["implementation_priority_score"], row["canonical_id"]),
|
|
)
|
|
usage_target.write_text(render_usage_pack(family, family_rows), encoding="utf-8")
|
|
roadmap = ROOT / "docs" / "research" / "technique_integration_roadmap_latest.md"
|
|
roadmap.parent.mkdir(parents=True, exist_ok=True)
|
|
roadmap.write_text(render_roadmap(report), encoding="utf-8")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--write", action="store_true")
|
|
parser.add_argument("--check-web", action="store_true")
|
|
parser.add_argument("--output", default="docs/research/technique_gap_scan_latest.json")
|
|
args = parser.parse_args()
|
|
report = scan()
|
|
if args.check_web:
|
|
report["web_receipt"] = check_web()
|
|
if args.write:
|
|
write_outputs(report, ROOT / args.output)
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|