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