| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | PPT Master - Page Context Projection |
| 4 | |
| 5 | Build deterministic per-page execution views and optional token telemetry. |
| 6 | |
| 7 | Usage: |
| 8 | Imported by project_management.cli. |
| 9 | |
| 10 | Examples: |
| 11 | build_page_context(Path("projects/demo"), "P07") |
| 12 | |
| 13 | Dependencies: |
| 14 | None for projection; tiktoken is optional for exact usage counts. |
| 15 | """ |
| 16 | |
| 17 | from __future__ import annotations |
| 18 | |
| 19 | import hashlib |
| 20 | import json |
| 21 | import math |
| 22 | import re |
| 23 | import statistics |
| 24 | import xml.etree.ElementTree as ET |
| 25 | from dataclasses import dataclass |
| 26 | from pathlib import Path |
| 27 | from typing import Callable, Iterable |
| 28 | |
| 29 | from .paths import SKILL_DIR as _SKILL_DIR |
| 30 | from .project_specs import ( |
| 31 | default_spec_lock_forbidden, |
| 32 | parse_markdown_artifact, |
| 33 | parse_spec_lock_artifact, |
| 34 | validate_project_artifacts, |
| 35 | ) |
| 36 | from svg_to_pptx.pptx_package.template_structure import ( |
| 37 | PptxStructureLock, |
| 38 | TemplateStructureError, |
| 39 | load_pptx_structure_lock, |
| 40 | ) |
| 41 | from visualization_catalog import ( |
| 42 | LEGACY_STRUCTURE_INTENT_KIND, |
| 43 | VISUALIZATION_SVG_KIND, |
| 44 | VisualizationCatalogError, |
| 45 | VisualizationEntry, |
| 46 | resolve_visualization_reference, |
| 47 | ) |
| 48 | |
| 49 | |
| 50 | PAGE_CONTEXT_SCHEMA = "ppt-master.page-context.v2" |
| 51 | PAGE_CONTEXT_USAGE_SCHEMA = "ppt-master.page-context-usage.v2" |
| 52 | PAGE_CONTEXT_REPORT_SCHEMA = "ppt-master.page-context-usage-report.v2" |
| 53 | TOKEN_ENCODING = "o200k_base" |
| 54 | PAGE_CONTEXT_TOKEN_TARGET = 2000 |
| 55 | LOCK_PROJECTION_TOKEN_TARGET = 1000 |
| 56 | |
| 57 | _PAGE_RE = re.compile(r"^(?:P)?([0-9]+)$", re.IGNORECASE) |
| 58 | _SLIDE_HEADING_RE = re.compile( |
| 59 | r"^#{3,6}[ \t]+Slide[ \t]+0*([0-9]+)(?:[ \t]*(?:[-:–—]).*)?$", |
| 60 | re.IGNORECASE | re.MULTILINE, |
| 61 | ) |
| 62 | _BLOCK_BOUNDARY_RE = re.compile(r"^#{2,6}[ \t]+", re.MULTILINE) |
| 63 | _PART_HEADING_RE = re.compile(r"^###[ \t]+(?!#)(.+?)[ \t]*$", re.MULTILINE) |
| 64 | _PAGE_TOKEN_RE = re.compile( |
| 65 | r"(?<![A-Za-z0-9_])P0*([1-9][0-9]*)(?![A-Za-z0-9_])", |
| 66 | re.IGNORECASE, |
| 67 | ) |
| 68 | |
| 69 | |
| 70 | class PageContextError(RuntimeError): |
| 71 | """Reject an incomplete or ambiguous page-context request.""" |
| 72 | |
| 73 | |
| 74 | @dataclass(frozen=True) |
| 75 | class PageRead: |
| 76 | """One exact model-visible page payload.""" |
| 77 | |
| 78 | kind: str |
| 79 | path: str |
| 80 | payload: str |
| 81 | |
| 82 | |
| 83 | @dataclass(frozen=True) |
| 84 | class PageContextResult: |
| 85 | """One projected page view plus the files that make it current.""" |
| 86 | |
| 87 | project_path: Path |
| 88 | page: str |
| 89 | context: dict[str, object] |
| 90 | inputs: tuple[Path, ...] |
| 91 | |
| 92 | |
| 93 | def normalize_page_key(raw_page: str) -> tuple[str, int]: |
| 94 | """Normalize a positive page identifier to the schema's P<NN> form.""" |
| 95 | match = _PAGE_RE.fullmatch(raw_page.strip()) |
| 96 | if match is None or int(match.group(1)) <= 0: |
| 97 | raise PageContextError("page must be a positive P<NN> identifier") |
| 98 | number = int(match.group(1)) |
| 99 | return f"P{number:02d}", number |
| 100 | |
| 101 | |
| 102 | def _section_index( |
| 103 | sections: Iterable[dict[str, object]], |
| 104 | ) -> dict[str, dict[str, object]]: |
| 105 | return { |
| 106 | str(section["heading"]).strip().casefold(): section |
| 107 | for section in sections |
| 108 | } |
| 109 | |
| 110 | |
| 111 | def _section_fields( |
| 112 | sections: dict[str, dict[str, object]], |
| 113 | heading: str, |
| 114 | ) -> dict[str, str]: |
| 115 | section = sections.get(heading.casefold()) |
| 116 | if section is None: |
| 117 | return {} |
| 118 | fields = section.get("fields", {}) |
| 119 | if not isinstance(fields, dict): |
| 120 | return {} |
| 121 | return {str(key): str(value) for key, value in fields.items()} |
| 122 | |
| 123 | |
| 124 | def _forbidden_items( |
| 125 | sections: dict[str, dict[str, object]], |
| 126 | ) -> list[str]: |
| 127 | section = sections.get("forbidden") |
| 128 | if section is None: |
| 129 | return [] |
| 130 | items: list[str] = [] |
| 131 | default_items = default_spec_lock_forbidden() |
| 132 | for raw_line in str(section.get("body", "")).splitlines(): |
| 133 | line = raw_line.strip() |
| 134 | if not line: |
| 135 | continue |
| 136 | item = re.sub(r"^-[ \t]+", "", line) |
| 137 | if item not in default_items: |
| 138 | items.append(item) |
| 139 | return items |
| 140 | |
| 141 | |
| 142 | def _outline_section( |
| 143 | sections: Iterable[dict[str, object]], |
| 144 | ) -> dict[str, object] | None: |
| 145 | for section in sections: |
| 146 | heading = str(section.get("heading", "")).strip().casefold() |
| 147 | if heading == "content outline" or heading.endswith(". content outline"): |
| 148 | return section |
| 149 | return None |
| 150 | |
| 151 | |
| 152 | def _page_image_filenames( |
| 153 | design_sections: Iterable[dict[str, object]], |
| 154 | page_number: int, |
| 155 | ) -> tuple[set[str], set[str]]: |
| 156 | """Read explicit P<NN> usage from the canonical image-resource table.""" |
| 157 | section = next( |
| 158 | ( |
| 159 | item |
| 160 | for item in design_sections |
| 161 | if ( |
| 162 | (heading := str(item.get("heading", "")).strip().casefold()) |
| 163 | == "image resource list" |
| 164 | or heading.startswith("viii. image resource list") |
| 165 | ) |
| 166 | ), |
| 167 | None, |
| 168 | ) |
| 169 | if section is None: |
| 170 | return set(), set() |
| 171 | table_rows = [ |
| 172 | [ |
| 173 | cell.strip().replace(r"\|", "|") |
| 174 | for cell in re.split(r"(?<!\\)\|", line.strip().strip("|")) |
| 175 | ] |
| 176 | for line in str(section.get("body", "")).splitlines() |
| 177 | if line.strip().startswith("|") and line.strip().endswith("|") |
| 178 | ] |
| 179 | if not table_rows: |
| 180 | return set(), set() |
| 181 | header = { |
| 182 | name.casefold(): index |
| 183 | for index, name in enumerate(table_rows[0]) |
| 184 | } |
| 185 | filename_index = header.get("filename") |
| 186 | purpose_index = header.get("purpose") |
| 187 | if filename_index is None or purpose_index is None: |
| 188 | return set(), set() |
| 189 | assigned: set[str] = set() |
| 190 | selected: set[str] = set() |
| 191 | for row in table_rows[1:]: |
| 192 | if len(row) <= max(filename_index, purpose_index): |
| 193 | continue |
| 194 | purpose = row[purpose_index] |
| 195 | pages = {int(match.group(1)) for match in _PAGE_TOKEN_RE.finditer(purpose)} |
| 196 | if not pages: |
| 197 | continue |
| 198 | filename = row[filename_index].strip().strip("`") |
| 199 | if not filename: |
| 200 | continue |
| 201 | basename = Path(filename).name |
| 202 | assigned.add(basename) |
| 203 | if page_number in pages: |
| 204 | selected.add(basename) |
| 205 | return assigned, selected |
| 206 | |
| 207 | |
| 208 | def _locked_image_basename(value: str) -> str: |
| 209 | return Path(value.split("|", 1)[0].strip()).name |
| 210 | |
| 211 | |
| 212 | def _outline_image_assignments( |
| 213 | design_sections: Iterable[dict[str, object]], |
| 214 | locked_images: dict[str, str], |
| 215 | ) -> set[str]: |
| 216 | outline = _outline_section(design_sections) |
| 217 | if outline is None: |
| 218 | return set() |
| 219 | body = str(outline.get("body", "")) |
| 220 | return { |
| 221 | _locked_image_basename(value) |
| 222 | for key, value in locked_images.items() |
| 223 | if any( |
| 224 | _contains_token(body, token) |
| 225 | for token in ( |
| 226 | key, |
| 227 | value.split("|", 1)[0].strip(), |
| 228 | _locked_image_basename(value), |
| 229 | ) |
| 230 | ) |
| 231 | } |
| 232 | |
| 233 | |
| 234 | def _slide_block( |
| 235 | design_sections: Iterable[dict[str, object]], |
| 236 | page_number: int, |
| 237 | ) -> tuple[str | None, str]: |
| 238 | outline = _outline_section(design_sections) |
| 239 | if outline is None: |
| 240 | raise PageContextError("design_spec.md has no Content Outline section") |
| 241 | body = str(outline.get("body", "")) |
| 242 | matches = [ |
| 243 | match |
| 244 | for match in _SLIDE_HEADING_RE.finditer(body) |
| 245 | if int(match.group(1)) == page_number |
| 246 | ] |
| 247 | if not matches: |
| 248 | raise PageContextError( |
| 249 | f"design_spec.md Content Outline has no Slide {page_number:02d} block" |
| 250 | ) |
| 251 | if len(matches) > 1: |
| 252 | raise PageContextError( |
| 253 | f"design_spec.md Content Outline repeats Slide {page_number:02d}" |
| 254 | ) |
| 255 | match = matches[0] |
| 256 | next_boundary = _BLOCK_BOUNDARY_RE.search(body, match.end()) |
| 257 | block_end = next_boundary.start() if next_boundary else len(body) |
| 258 | block = body[match.start():block_end].strip() |
| 259 | part_matches = list(_PART_HEADING_RE.finditer(body, 0, match.start())) |
| 260 | part = part_matches[-1].group(1).strip() if part_matches else None |
| 261 | return part, block |
| 262 | |
| 263 | |
| 264 | def _relative_project_path(project_path: Path, path: Path) -> str: |
| 265 | try: |
| 266 | return path.resolve().relative_to(project_path).as_posix() |
| 267 | except ValueError as exc: |
| 268 | raise PageContextError(f"path escapes project: {path}") from exc |
| 269 | |
| 270 | |
| 271 | def _prototype_image_refs(svg_path: Path) -> list[str]: |
| 272 | try: |
| 273 | root = ET.parse(svg_path).getroot() |
| 274 | except (OSError, ET.ParseError) as exc: |
| 275 | raise PageContextError(f"cannot read prototype SVG {svg_path}: {exc}") from exc |
| 276 | refs: set[str] = set() |
| 277 | for element in root.iter(): |
| 278 | if element.tag.rsplit("}", 1)[-1] != "image": |
| 279 | continue |
| 280 | for name, value in element.attrib.items(): |
| 281 | if name.rsplit("}", 1)[-1] != "href": |
| 282 | continue |
| 283 | normalized = value.strip() |
| 284 | if normalized and not normalized.startswith(("data:", "#")): |
| 285 | refs.add(normalized) |
| 286 | return sorted(refs) |
| 287 | |
| 288 | |
| 289 | def _contains_token(text: str, token: str) -> bool: |
| 290 | if not token: |
| 291 | return False |
| 292 | if re.fullmatch(r"[A-Za-z0-9_]+", token): |
| 293 | return re.search( |
| 294 | rf"(?<![A-Za-z0-9_]){re.escape(token)}(?![A-Za-z0-9_])", |
| 295 | text, |
| 296 | ) is not None |
| 297 | return token in text |
| 298 | |
| 299 | |
| 300 | def _page_images( |
| 301 | locked_images: dict[str, str], |
| 302 | brief: str, |
| 303 | prototype_refs: list[str], |
| 304 | assigned_filenames: set[str], |
| 305 | resolved_filenames: set[str], |
| 306 | ) -> tuple[str, dict[str, str]]: |
| 307 | if not locked_images: |
| 308 | return "none", {} |
| 309 | ref_basenames = {Path(ref).name for ref in prototype_refs} |
| 310 | selected: dict[str, str] = {} |
| 311 | unresolved: dict[str, str] = {} |
| 312 | for key, value in locked_images.items(): |
| 313 | basename = _locked_image_basename(value) |
| 314 | if ( |
| 315 | _contains_token(brief, key) |
| 316 | or _contains_token(brief, value) |
| 317 | or _contains_token(brief, basename) |
| 318 | or basename in ref_basenames |
| 319 | or basename in assigned_filenames |
| 320 | ): |
| 321 | selected[key] = value |
| 322 | elif basename not in resolved_filenames: |
| 323 | unresolved[key] = value |
| 324 | if selected and unresolved: |
| 325 | return "explicit+unassigned", {**selected, **unresolved} |
| 326 | if selected: |
| 327 | return "explicit", selected |
| 328 | if unresolved: |
| 329 | return "unassigned", unresolved |
| 330 | return "confirmed-none", {} |
| 331 | |
| 332 | |
| 333 | def _page_template( |
| 334 | project_path: Path, |
| 335 | structure_lock: PptxStructureLock | None, |
| 336 | page_number: int, |
| 337 | ) -> tuple[dict[str, object] | None, Path | None]: |
| 338 | if structure_lock is None or structure_lock.mode != "structured": |
| 339 | return None, None |
| 340 | prototype = next( |
| 341 | (item for item in structure_lock.prototypes if item.slide_num == page_number), |
| 342 | None, |
| 343 | ) |
| 344 | assignment = next( |
| 345 | (item for item in structure_lock.layouts if item.slide_num == page_number), |
| 346 | None, |
| 347 | ) |
| 348 | if prototype is None or assignment is None: |
| 349 | raise PageContextError( |
| 350 | f"structured lock has no complete mapping for P{page_number:02d}" |
| 351 | ) |
| 352 | definition = next( |
| 353 | ( |
| 354 | item |
| 355 | for item in structure_lock.layout_definitions |
| 356 | if item.layout_key == assignment.layout_key |
| 357 | ), |
| 358 | None, |
| 359 | ) |
| 360 | if definition is None: |
| 361 | raise PageContextError( |
| 362 | f"structured lock has no definition for Layout {assignment.layout_key!r}" |
| 363 | ) |
| 364 | master = next( |
| 365 | ( |
| 366 | item |
| 367 | for item in structure_lock.masters |
| 368 | if item.master_key == definition.master_key |
| 369 | ), |
| 370 | None, |
| 371 | ) |
| 372 | if master is None: |
| 373 | raise PageContextError( |
| 374 | f"structured lock has no definition for Master {definition.master_key!r}" |
| 375 | ) |
| 376 | template = { |
| 377 | "reuse_scope": structure_lock.template_reuse_scope, |
| 378 | "adherence": structure_lock.template_adherence, |
| 379 | "prototype": prototype.template_basename, |
| 380 | "prototype_path": _relative_project_path(project_path, prototype.svg_path), |
| 381 | "layout": { |
| 382 | "key": definition.layout_key, |
| 383 | "name": definition.layout_name, |
| 384 | "source": ( |
| 385 | f"P{definition.prototype_slide_num:02d}" |
| 386 | if definition.prototype_slide_num is not None |
| 387 | else _relative_project_path( |
| 388 | project_path, |
| 389 | definition.prototype_svg_path, |
| 390 | ) |
| 391 | ), |
| 392 | }, |
| 393 | "master": { |
| 394 | "key": master.master_key, |
| 395 | "name": master.master_name, |
| 396 | }, |
| 397 | } |
| 398 | return template, prototype.svg_path |
| 399 | |
| 400 | |
| 401 | def _reference_payload( |
| 402 | kind: str, |
| 403 | path: Path, |
| 404 | *, |
| 405 | scope: str, |
| 406 | display_path: str, |
| 407 | same_context_edit_policy: str | None = None, |
| 408 | ) -> dict[str, str]: |
| 409 | """Describe one large reference without injecting its contents per page.""" |
| 410 | payload = { |
| 411 | "kind": kind, |
| 412 | "scope": scope, |
| 413 | "path": display_path, |
| 414 | "sha256": _file_sha256(path), |
| 415 | "load_policy": "once-per-execution-context", |
| 416 | } |
| 417 | if same_context_edit_policy is not None: |
| 418 | payload["same_context_edit_policy"] = same_context_edit_policy |
| 419 | return payload |
| 420 | |
| 421 | |
| 422 | def _visualization_reference( |
| 423 | value: str, |
| 424 | *, |
| 425 | allow_legacy_bare: bool, |
| 426 | ) -> tuple[dict[str, str] | None, Path | None, VisualizationEntry]: |
| 427 | """Resolve one live asset or one legacy intent-only Structure key.""" |
| 428 | source_section = "page_charts" if allow_legacy_bare else "page_visualizations" |
| 429 | try: |
| 430 | entry = resolve_visualization_reference( |
| 431 | value, |
| 432 | allow_legacy_bare=allow_legacy_bare, |
| 433 | ) |
| 434 | except VisualizationCatalogError as exc: |
| 435 | raise PageContextError( |
| 436 | f"{source_section} value {value!r} cannot resolve a visualization: {exc}" |
| 437 | ) from exc |
| 438 | if entry.kind == LEGACY_STRUCTURE_INTENT_KIND: |
| 439 | if not allow_legacy_bare or entry.path is not None: |
| 440 | raise PageContextError( |
| 441 | f"{source_section} value {value!r} has an invalid legacy " |
| 442 | "Structure intent resolution" |
| 443 | ) |
| 444 | return None, None, entry |
| 445 | if entry.kind != VISUALIZATION_SVG_KIND or entry.path is None: |
| 446 | raise PageContextError( |
| 447 | f"{source_section} value {value!r} resolves to unsupported kind " |
| 448 | f"{entry.kind!r}" |
| 449 | ) |
| 450 | path = Path(entry.path).resolve() |
| 451 | try: |
| 452 | display_path = path.relative_to(_SKILL_DIR.resolve()).as_posix() |
| 453 | except ValueError as exc: |
| 454 | raise PageContextError( |
| 455 | f"{source_section} value {value!r} resolves outside the Skill: {path}" |
| 456 | ) from exc |
| 457 | payload = _reference_payload( |
| 458 | entry.kind, |
| 459 | path, |
| 460 | scope="skill", |
| 461 | display_path=display_path, |
| 462 | ) |
| 463 | payload.update( |
| 464 | { |
| 465 | "reference": entry.reference, |
| 466 | "family": entry.family, |
| 467 | "key": entry.key, |
| 468 | } |
| 469 | ) |
| 470 | return payload, path, entry |
| 471 | |
| 472 | |
| 473 | def build_page_context(project: str | Path, raw_page: str) -> PageContextResult: |
| 474 | """Build one current per-page projection without writing the project.""" |
| 475 | project_path = Path(project).resolve() |
| 476 | if not project_path.is_dir(): |
| 477 | raise PageContextError(f"project directory not found: {project_path}") |
| 478 | page, page_number = normalize_page_key(raw_page) |
| 479 | lock_path = project_path / "spec_lock.md" |
| 480 | design_path = project_path / "design_spec.md" |
| 481 | for required in (lock_path, design_path): |
| 482 | if not required.is_file(): |
| 483 | raise PageContextError(f"required artifact not found: {required.name}") |
| 484 | preflight_errors, _preflight_warnings = validate_project_artifacts( |
| 485 | project_path, |
| 486 | include_design=False, |
| 487 | ) |
| 488 | if preflight_errors: |
| 489 | preview = "; ".join(preflight_errors[:8]) |
| 490 | suffix = ( |
| 491 | "" |
| 492 | if len(preflight_errors) <= 8 |
| 493 | else f"; +{len(preflight_errors) - 8} more" |
| 494 | ) |
| 495 | raise PageContextError( |
| 496 | "spec_lock/template preflight failed before page generation: " |
| 497 | f"{preview}{suffix}" |
| 498 | ) |
| 499 | try: |
| 500 | lock_sections_raw = parse_spec_lock_artifact( |
| 501 | lock_path, |
| 502 | report_duplicate_fields=True, |
| 503 | ) |
| 504 | design_sections = parse_markdown_artifact(design_path) |
| 505 | except (OSError, ValueError) as exc: |
| 506 | raise PageContextError(str(exc)) from exc |
| 507 | lock_sections = _section_index(lock_sections_raw) |
| 508 | part, brief = _slide_block(design_sections, page_number) |
| 509 | warnings: list[str] = [] |
| 510 | rhythm_fields = _section_fields(lock_sections, "page_rhythm") |
| 511 | rhythm = rhythm_fields.get(page) |
| 512 | if rhythm is None: |
| 513 | rhythm = "dense" |
| 514 | warnings.append(f"page_rhythm has no {page}; using compatibility default dense") |
| 515 | visualization_value = _section_fields( |
| 516 | lock_sections, |
| 517 | "page_visualizations", |
| 518 | ).get(page) |
| 519 | legacy_chart_key = _section_fields(lock_sections, "page_charts").get(page) |
| 520 | if visualization_value is not None and legacy_chart_key is not None: |
| 521 | raise PageContextError( |
| 522 | f"{page} is declared in both page_visualizations and legacy " |
| 523 | "page_charts; keep only page_visualizations" |
| 524 | ) |
| 525 | try: |
| 526 | structure_lock = load_pptx_structure_lock(project_path) |
| 527 | except TemplateStructureError as exc: |
| 528 | raise PageContextError(str(exc)) from exc |
| 529 | template, prototype_path = _page_template( |
| 530 | project_path, |
| 531 | structure_lock, |
| 532 | page_number, |
| 533 | ) |
| 534 | prototype_refs = ( |
| 535 | _prototype_image_refs(prototype_path) |
| 536 | if prototype_path is not None |
| 537 | else [] |
| 538 | ) |
| 539 | table_assigned_filenames, assigned_filenames = _page_image_filenames( |
| 540 | design_sections, |
| 541 | page_number, |
| 542 | ) |
| 543 | locked_images = _section_fields(lock_sections, "images") |
| 544 | resolved_filenames = ( |
| 545 | {_locked_image_basename(value) for value in locked_images.values()} |
| 546 | if structure_lock is not None |
| 547 | and structure_lock.template_reuse_scope == "mirror" |
| 548 | else table_assigned_filenames |
| 549 | | _outline_image_assignments(design_sections, locked_images) |
| 550 | ) |
| 551 | image_selection, selected_images = _page_images( |
| 552 | locked_images, |
| 553 | brief, |
| 554 | ( |
| 555 | prototype_refs |
| 556 | if structure_lock is not None |
| 557 | and structure_lock.template_reuse_scope == "mirror" |
| 558 | else [] |
| 559 | ), |
| 560 | assigned_filenames, |
| 561 | resolved_filenames, |
| 562 | ) |
| 563 | inputs = [lock_path, design_path] |
| 564 | reference_set: list[dict[str, str]] = [ |
| 565 | _reference_payload( |
| 566 | "design-spec", |
| 567 | design_path, |
| 568 | scope="project", |
| 569 | display_path="design_spec.md", |
| 570 | same_context_edit_policy="targeted-readback-and-rebind", |
| 571 | ), |
| 572 | ] |
| 573 | template_design_path = project_path / "templates" / "design_spec.md" |
| 574 | if template_design_path.is_file(): |
| 575 | inputs.append(template_design_path) |
| 576 | reference_set.append( |
| 577 | _reference_payload( |
| 578 | "template-design-spec", |
| 579 | template_design_path, |
| 580 | scope="project", |
| 581 | display_path="templates/design_spec.md", |
| 582 | ) |
| 583 | ) |
| 584 | if prototype_path is not None: |
| 585 | inputs.append(prototype_path) |
| 586 | reference_set.append( |
| 587 | _reference_payload( |
| 588 | "prototype-svg", |
| 589 | prototype_path, |
| 590 | scope="project", |
| 591 | display_path=_relative_project_path(project_path, prototype_path), |
| 592 | ) |
| 593 | ) |
| 594 | visualization_entry: VisualizationEntry | None = None |
| 595 | selected_visualization = ( |
| 596 | visualization_value |
| 597 | if visualization_value is not None |
| 598 | else legacy_chart_key |
| 599 | ) |
| 600 | if selected_visualization is not None: |
| 601 | visualization_reference, visualization_path, visualization_entry = ( |
| 602 | _visualization_reference( |
| 603 | selected_visualization, |
| 604 | allow_legacy_bare=visualization_value is None, |
| 605 | ) |
| 606 | ) |
| 607 | if visualization_path is not None and visualization_reference is not None: |
| 608 | inputs.append(visualization_path) |
| 609 | reference_set.append(visualization_reference) |
| 610 | mode_fields = _section_fields(lock_sections, "mode") |
| 611 | visual_style_fields = _section_fields(lock_sections, "visual_style") |
| 612 | # Each on-demand projection includes bounded lock anchors; large reference |
| 613 | # payloads stay outside it and are represented by reference_set. |
| 614 | global_context = { |
| 615 | "communication": _section_fields(lock_sections, "communication"), |
| 616 | "canvas": _section_fields(lock_sections, "canvas"), |
| 617 | "mode": mode_fields.get("mode"), |
| 618 | "mode_behavior": mode_fields.get("mode_behavior"), |
| 619 | "visual_style": visual_style_fields.get("visual_style"), |
| 620 | "visual_style_behavior": visual_style_fields.get( |
| 621 | "visual_style_behavior" |
| 622 | ), |
| 623 | "colors": _section_fields(lock_sections, "colors"), |
| 624 | "typography": _section_fields(lock_sections, "typography"), |
| 625 | "icons": _section_fields(lock_sections, "icons"), |
| 626 | "pptx_structure": _section_fields(lock_sections, "pptx_structure"), |
| 627 | "forbidden": _forbidden_items(lock_sections), |
| 628 | } |
| 629 | global_context = { |
| 630 | key: value |
| 631 | for key, value in global_context.items() |
| 632 | if value not in ({}, [], None, "") |
| 633 | } |
| 634 | current_page: dict[str, object] = { |
| 635 | "part": part, |
| 636 | "brief_markdown": brief, |
| 637 | "rhythm": rhythm, |
| 638 | "image_selection": image_selection, |
| 639 | } |
| 640 | if ( |
| 641 | visualization_entry is not None |
| 642 | and visualization_entry.kind == VISUALIZATION_SVG_KIND |
| 643 | ): |
| 644 | current_page["visualization"] = visualization_entry.reference |
| 645 | elif ( |
| 646 | visualization_entry is not None |
| 647 | and visualization_entry.kind == LEGACY_STRUCTURE_INTENT_KIND |
| 648 | ): |
| 649 | current_page["structure_intent"] = visualization_entry.key |
| 650 | if legacy_chart_key is not None: |
| 651 | current_page["chart"] = legacy_chart_key |
| 652 | if selected_images: |
| 653 | current_page["images"] = selected_images |
| 654 | if template is not None: |
| 655 | current_page["template"] = template |
| 656 | context: dict[str, object] = { |
| 657 | "schema": PAGE_CONTEXT_SCHEMA, |
| 658 | "page": page, |
| 659 | "lock_source": { |
| 660 | "path": "spec_lock.md", |
| 661 | "sha256": _file_sha256(lock_path), |
| 662 | "load_policy": "on-demand-anchor-projection", |
| 663 | }, |
| 664 | "global": global_context, |
| 665 | "page_context": current_page, |
| 666 | "reference_set": reference_set, |
| 667 | } |
| 668 | if warnings: |
| 669 | context["warnings"] = warnings |
| 670 | unique_inputs = tuple(dict.fromkeys(path.resolve() for path in inputs)) |
| 671 | return PageContextResult( |
| 672 | project_path=project_path, |
| 673 | page=page, |
| 674 | context=context, |
| 675 | inputs=unique_inputs, |
| 676 | ) |
| 677 | |
| 678 | |
| 679 | def _compact_json(payload: object) -> str: |
| 680 | return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n" |
| 681 | |
| 682 | |
| 683 | def _pretty_json(payload: object) -> str: |
| 684 | return json.dumps(payload, ensure_ascii=False, indent=2) + "\n" |
| 685 | |
| 686 | |
| 687 | def render_page_context( |
| 688 | result: PageContextResult, |
| 689 | *, |
| 690 | bundle: bool = False, |
| 691 | pretty: bool = False, |
| 692 | ) -> tuple[str, tuple[PageRead, ...]]: |
| 693 | """Render compact stdout; ``bundle`` remains a compatibility no-op.""" |
| 694 | context_payload = ( |
| 695 | _pretty_json(result.context) if pretty else _compact_json(result.context) |
| 696 | ) |
| 697 | context_read = PageRead( |
| 698 | kind="page-context", |
| 699 | path="stdout:page-context", |
| 700 | payload=context_payload, |
| 701 | ) |
| 702 | return context_payload, (context_read,) |
| 703 | |
| 704 | |
| 705 | def _sha256_bytes(payload: bytes) -> str: |
| 706 | return hashlib.sha256(payload).hexdigest() |
| 707 | |
| 708 | |
| 709 | def _file_sha256(path: Path) -> str: |
| 710 | digest = hashlib.sha256() |
| 711 | with path.open("rb") as stream: |
| 712 | for chunk in iter(lambda: stream.read(1024 * 1024), b""): |
| 713 | digest.update(chunk) |
| 714 | return digest.hexdigest() |
| 715 | |
| 716 | |
| 717 | def _input_location(project_path: Path, path: Path) -> tuple[str, str]: |
| 718 | """Return a stable project- or Skill-relative locator for telemetry.""" |
| 719 | resolved = path.resolve() |
| 720 | for scope, root in (("project", project_path), ("skill", _SKILL_DIR)): |
| 721 | try: |
| 722 | return scope, resolved.relative_to(root.resolve()).as_posix() |
| 723 | except ValueError: |
| 724 | continue |
| 725 | raise PageContextError(f"input escapes project and Skill roots: {path}") |
| 726 | |
| 727 | |
| 728 | def _resolve_input_location( |
| 729 | project_path: Path, |
| 730 | scope: str, |
| 731 | relative_path: str, |
| 732 | ) -> Path | None: |
| 733 | """Resolve one recorded input without accepting arbitrary filesystem roots.""" |
| 734 | roots = {"project": project_path, "skill": _SKILL_DIR} |
| 735 | root = roots.get(scope) |
| 736 | if root is None: |
| 737 | return None |
| 738 | resolved = (root / relative_path).resolve() |
| 739 | try: |
| 740 | resolved.relative_to(root.resolve()) |
| 741 | except ValueError: |
| 742 | return None |
| 743 | return resolved |
| 744 | |
| 745 | |
| 746 | def _token_counter() -> tuple[Callable[[str], int] | None, str]: |
| 747 | try: |
| 748 | import tiktoken |
| 749 | except ImportError: |
| 750 | return None, "unavailable" |
| 751 | try: |
| 752 | encoder = tiktoken.get_encoding(TOKEN_ENCODING) |
| 753 | except Exception: |
| 754 | return None, "unavailable" |
| 755 | return ( |
| 756 | lambda text: len(encoder.encode(text, disallowed_special=())), |
| 757 | "exact", |
| 758 | ) |
| 759 | |
| 760 | |
| 761 | def _payload_measurement( |
| 762 | read: PageRead, |
| 763 | count_tokens: Callable[[str], int] | None, |
| 764 | ) -> dict[str, object]: |
| 765 | payload = read.payload.encode("utf-8") |
| 766 | measurement: dict[str, object] = { |
| 767 | "kind": read.kind, |
| 768 | "scope": "component" if read.kind == "lock-projection" else "page", |
| 769 | "path": read.path, |
| 770 | "sha256": _sha256_bytes(payload), |
| 771 | "utf8_bytes": len(payload), |
| 772 | "characters": len(read.payload), |
| 773 | "tokens": count_tokens(read.payload) if count_tokens else None, |
| 774 | } |
| 775 | return measurement |
| 776 | |
| 777 | |
| 778 | def record_page_context_usage( |
| 779 | result: PageContextResult, |
| 780 | output: str, |
| 781 | measured_reads: tuple[PageRead, ...], |
| 782 | ) -> tuple[Path, str]: |
| 783 | """Write one deterministic, derived token snapshot for the current page.""" |
| 784 | count_tokens, token_status = _token_counter() |
| 785 | lock_read = PageRead( |
| 786 | kind="lock-projection", |
| 787 | path="stdout:global", |
| 788 | payload=_compact_json(result.context["global"]), |
| 789 | ) |
| 790 | documents = [ |
| 791 | _payload_measurement(read, count_tokens) |
| 792 | for read in (*measured_reads, lock_read) |
| 793 | ] |
| 794 | output_bytes = output.encode("utf-8") |
| 795 | input_records: list[dict[str, object]] = [] |
| 796 | for path in result.inputs: |
| 797 | scope, relative_path = _input_location(result.project_path, path) |
| 798 | input_records.append( |
| 799 | { |
| 800 | "scope": scope, |
| 801 | "path": relative_path, |
| 802 | "exists": True, |
| 803 | "sha256": _file_sha256(path), |
| 804 | } |
| 805 | ) |
| 806 | by_kind = { |
| 807 | str(item["kind"]): item.get("tokens") |
| 808 | for item in documents |
| 809 | } |
| 810 | route = dict(result.context["global"].get("pptx_structure", {})) |
| 811 | template = result.context["page_context"].get("template") |
| 812 | if isinstance(template, dict): |
| 813 | if isinstance(value := template.get("reuse_scope"), str): |
| 814 | route["template_reuse_scope"] = value |
| 815 | usage = { |
| 816 | "schema": PAGE_CONTEXT_USAGE_SCHEMA, |
| 817 | "page": result.page, |
| 818 | "output_mode": "compact", |
| 819 | "route": route, |
| 820 | "encoding": TOKEN_ENCODING, |
| 821 | "token_status": token_status, |
| 822 | "image_selection": result.context["page_context"]["image_selection"], |
| 823 | "inputs": input_records, |
| 824 | "references": result.context.get("reference_set", []), |
| 825 | "documents": documents, |
| 826 | "controlled_output": { |
| 827 | "sha256": _sha256_bytes(output_bytes), |
| 828 | "utf8_bytes": len(output_bytes), |
| 829 | "characters": len(output), |
| 830 | "tokens": count_tokens(output) if count_tokens else None, |
| 831 | }, |
| 832 | "totals": { |
| 833 | "page_context": by_kind.get("page-context"), |
| 834 | "lock_projection": by_kind.get("lock-projection"), |
| 835 | }, |
| 836 | "targets": { |
| 837 | "page_context_max_tokens": PAGE_CONTEXT_TOKEN_TARGET, |
| 838 | "lock_projection_max_tokens": LOCK_PROJECTION_TOKEN_TARGET, |
| 839 | }, |
| 840 | "untracked": [ |
| 841 | "source-material reads", |
| 842 | "once-per-execution-context reference payloads", |
| 843 | "other session-level prompt references", |
| 844 | ], |
| 845 | } |
| 846 | usage_dir = result.project_path / "analysis" / "page-context" |
| 847 | usage_dir.mkdir(parents=True, exist_ok=True) |
| 848 | usage_path = usage_dir / f"{result.page}.usage.json" |
| 849 | temporary_path = usage_path.with_suffix(".usage.json.tmp") |
| 850 | temporary_path.write_text(_pretty_json(usage), encoding="utf-8") |
| 851 | temporary_path.replace(usage_path) |
| 852 | return usage_path, token_status |
| 853 | |
| 854 | |
| 855 | def _nearest_rank(values: list[int], percentile: float) -> int: |
| 856 | rank = max(1, math.ceil(percentile * len(values))) |
| 857 | return sorted(values)[rank - 1] |
| 858 | |
| 859 | |
| 860 | def _metric(values: list[int], *, target: int | None = None) -> dict[str, object]: |
| 861 | if not values: |
| 862 | return { |
| 863 | "count": 0, |
| 864 | "sum": 0, |
| 865 | "min": None, |
| 866 | "p50": None, |
| 867 | "p95": None, |
| 868 | "max": None, |
| 869 | **({"over_target_count": 0, "target": target} if target else {}), |
| 870 | } |
| 871 | metric: dict[str, object] = { |
| 872 | "count": len(values), |
| 873 | "sum": sum(values), |
| 874 | "min": min(values), |
| 875 | "p50": round(statistics.median(values)), |
| 876 | "p95": _nearest_rank(values, 0.95), |
| 877 | "max": max(values), |
| 878 | } |
| 879 | if target is not None: |
| 880 | metric.update({ |
| 881 | "target": target, |
| 882 | "over_target_count": sum(value > target for value in values), |
| 883 | }) |
| 884 | return metric |
| 885 | |
| 886 | |
| 887 | def page_context_usage_report(project: str | Path) -> dict[str, object]: |
| 888 | """Summarize fresh per-page telemetry without changing recorded history.""" |
| 889 | project_path = Path(project).resolve() |
| 890 | usage_dir = project_path / "analysis" / "page-context" |
| 891 | records: list[dict[str, object]] = [] |
| 892 | stale_pages: list[str] = [] |
| 893 | unavailable_pages: list[str] = [] |
| 894 | if usage_dir.is_dir(): |
| 895 | for usage_path in sorted(usage_dir.glob("P*.usage.json")): |
| 896 | try: |
| 897 | record = json.loads(usage_path.read_text(encoding="utf-8")) |
| 898 | except (OSError, json.JSONDecodeError): |
| 899 | stale_pages.append(usage_path.stem.split(".", 1)[0]) |
| 900 | continue |
| 901 | page = str(record.get("page", usage_path.stem.split(".", 1)[0])) |
| 902 | if record.get("schema") != PAGE_CONTEXT_USAGE_SCHEMA: |
| 903 | stale_pages.append(page) |
| 904 | continue |
| 905 | if record.get("output_mode") != "compact": |
| 906 | stale_pages.append(page) |
| 907 | continue |
| 908 | stale = False |
| 909 | for item in record.get("inputs", []): |
| 910 | if not isinstance(item, dict): |
| 911 | stale = True |
| 912 | break |
| 913 | source_path = _resolve_input_location( |
| 914 | project_path, |
| 915 | str(item.get("scope", "project")), |
| 916 | str(item.get("path", "")), |
| 917 | ) |
| 918 | if source_path is None: |
| 919 | stale = True |
| 920 | break |
| 921 | expected_exists = item.get("exists", True) |
| 922 | if expected_exists is False: |
| 923 | if source_path.exists(): |
| 924 | stale = True |
| 925 | break |
| 926 | elif ( |
| 927 | not source_path.is_file() |
| 928 | or _file_sha256(source_path) != item.get("sha256") |
| 929 | ): |
| 930 | stale = True |
| 931 | break |
| 932 | if stale: |
| 933 | stale_pages.append(page) |
| 934 | continue |
| 935 | if record.get("token_status") != "exact": |
| 936 | unavailable_pages.append(page) |
| 937 | records.append(record) |
| 938 | |
| 939 | def tokens_for(kind: str) -> list[int]: |
| 940 | values: list[int] = [] |
| 941 | for record in records: |
| 942 | for document in record.get("documents", []): |
| 943 | if not isinstance(document, dict) or document.get("kind") != kind: |
| 944 | continue |
| 945 | value = document.get("tokens") |
| 946 | if isinstance(value, int): |
| 947 | values.append(value) |
| 948 | return values |
| 949 | |
| 950 | controlled = [ |
| 951 | value |
| 952 | for record in records |
| 953 | if isinstance( |
| 954 | value := record.get("controlled_output", {}).get("tokens"), |
| 955 | int, |
| 956 | ) |
| 957 | ] |
| 958 | unique_references = sorted({ |
| 959 | f"{reference.get('scope', 'project')}:{reference.get('path', '')}" |
| 960 | for record in records |
| 961 | for reference in record.get("references", []) |
| 962 | if isinstance(reference, dict) and reference.get("path") |
| 963 | }) |
| 964 | return { |
| 965 | "schema": PAGE_CONTEXT_REPORT_SCHEMA, |
| 966 | "project": project_path.name, |
| 967 | "record_count": len(records), |
| 968 | "pages": sorted(str(record["page"]) for record in records), |
| 969 | "stale_pages": sorted(set(stale_pages)), |
| 970 | "token_unavailable_pages": sorted(set(unavailable_pages)), |
| 971 | "unique_reference_count": len(unique_references), |
| 972 | "unique_references": unique_references, |
| 973 | "metrics": { |
| 974 | "page_context": _metric( |
| 975 | tokens_for("page-context"), |
| 976 | target=PAGE_CONTEXT_TOKEN_TARGET, |
| 977 | ), |
| 978 | "lock_projection": _metric( |
| 979 | tokens_for("lock-projection"), |
| 980 | target=LOCK_PROJECTION_TOKEN_TARGET, |
| 981 | ), |
| 982 | "controlled_output": _metric(controlled), |
| 983 | }, |
| 984 | } |
| 985 |