192 lines
8.2 KiB
Python
192 lines
8.2 KiB
Python
"""Bounded, file-backed author guidance for the CDSL workflow.
|
|
|
|
The corpus is deliberately non-authoritative: contracts, schemas, topology
|
|
tokens, and server preflight always remain the executable source of truth.
|
|
Loading errors return an empty selection so authoring continues with the
|
|
pre-guidance prompt instead of turning documentation into an availability
|
|
dependency.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.cad_agent.domain.state import TaskPhase
|
|
from app.cad_agent.ports import AuthorGuidanceSelection
|
|
|
|
|
|
_MIN_CHARS = 1_200
|
|
_MAX_CHARS = 6_000
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _Section:
|
|
section_id: str
|
|
title: str
|
|
priority: int
|
|
mandatory: bool
|
|
content: str
|
|
|
|
@property
|
|
def block(self) -> str:
|
|
return f"## {self.title}\n{self.content.strip()}"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _Corpus:
|
|
version: str
|
|
sections: dict[str, _Section]
|
|
phase_sections: dict[str, tuple[str, ...]]
|
|
repair_sections: tuple[str, ...]
|
|
final_sections: tuple[str, ...]
|
|
operation_sections: dict[str, tuple[str, ...]]
|
|
|
|
|
|
class FileAuthorGuidance:
|
|
"""Read and select the checked-in corpus deterministically.
|
|
|
|
Selection depends exclusively on workflow state and the runtime operation
|
|
registry. It intentionally receives neither the user's request nor image
|
|
observations, so it cannot become an implicit part-family classifier.
|
|
"""
|
|
|
|
def __init__(self, root: Path, *, enabled: bool = True, max_chars: int = 3_600) -> None:
|
|
self.root = root
|
|
self.enabled = enabled
|
|
self.max_chars = min(_MAX_CHARS, max(_MIN_CHARS, max_chars))
|
|
self._corpus: _Corpus | None = None
|
|
self._load_error = ""
|
|
|
|
def select(
|
|
self,
|
|
*,
|
|
phase: TaskPhase,
|
|
atomic_id: str,
|
|
repair_required: bool,
|
|
supported_atomic_ids: tuple[str, ...],
|
|
) -> AuthorGuidanceSelection:
|
|
if not self.enabled:
|
|
return AuthorGuidanceSelection(fallback_reason="guidance_disabled")
|
|
corpus = self._load()
|
|
if corpus is None:
|
|
return AuthorGuidanceSelection(fallback_reason=self._load_error or "guidance_unavailable")
|
|
supported = set(supported_atomic_ids)
|
|
if set(corpus.operation_sections) != supported:
|
|
return AuthorGuidanceSelection(fallback_reason="guidance_operation_coverage_mismatch")
|
|
if atomic_id and atomic_id not in supported:
|
|
return AuthorGuidanceSelection(fallback_reason="guidance_unknown_atomic_id")
|
|
|
|
if repair_required:
|
|
requested = list(corpus.repair_sections)
|
|
elif phase == TaskPhase.FINAL_VALIDATION:
|
|
requested = list(corpus.final_sections)
|
|
else:
|
|
requested = list(corpus.phase_sections.get(phase.value, corpus.phase_sections.get("DEFAULT", ())))
|
|
if atomic_id:
|
|
requested.extend(corpus.operation_sections[atomic_id])
|
|
requested = list(dict.fromkeys(requested))
|
|
if not requested:
|
|
return AuthorGuidanceSelection(fallback_reason="guidance_no_matching_sections")
|
|
|
|
mandatory = [section_id for section_id in requested if corpus.sections[section_id].mandatory]
|
|
optional = [section_id for section_id in requested if not corpus.sections[section_id].mandatory]
|
|
optional.sort(key=lambda section_id: (-corpus.sections[section_id].priority, requested.index(section_id)))
|
|
selected: list[str] = []
|
|
text = ""
|
|
for section_id in [*mandatory, *optional]:
|
|
block = corpus.sections[section_id].block
|
|
candidate = block if not text else f"{text}\n\n{block}"
|
|
if len(candidate) <= self.max_chars:
|
|
text = candidate
|
|
selected.append(section_id)
|
|
elif section_id in mandatory:
|
|
# Do not silently drop contract, datum, or operation guidance.
|
|
return AuthorGuidanceSelection(fallback_reason="guidance_required_sections_exceed_budget")
|
|
return AuthorGuidanceSelection(
|
|
version=corpus.version,
|
|
section_ids=tuple(selected),
|
|
content=text,
|
|
enabled=True,
|
|
)
|
|
|
|
def _load(self) -> _Corpus | None:
|
|
if self._corpus is not None:
|
|
return self._corpus
|
|
if self._load_error:
|
|
return None
|
|
try:
|
|
manifest_path = self.root / "manifest.json"
|
|
raw = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
if not isinstance(raw, dict):
|
|
raise ValueError("manifest is not an object")
|
|
version = raw.get("version")
|
|
if raw.get("schema_version") != "cdsl.author-guidance.manifest.v1" or not isinstance(version, str) or not version:
|
|
raise ValueError("manifest version is invalid")
|
|
raw_sections = raw.get("sections")
|
|
if not isinstance(raw_sections, list) or not raw_sections:
|
|
raise ValueError("manifest sections are invalid")
|
|
sections: dict[str, _Section] = {}
|
|
root = self.root.resolve()
|
|
for item in raw_sections:
|
|
if not isinstance(item, dict):
|
|
raise ValueError("section declaration is invalid")
|
|
section_id = item.get("id")
|
|
filename = item.get("file")
|
|
title = item.get("title")
|
|
priority = item.get("priority")
|
|
mandatory = item.get("mandatory", False)
|
|
if (
|
|
not isinstance(section_id, str) or not section_id
|
|
or not isinstance(filename, str) or not filename
|
|
or not isinstance(title, str) or not title
|
|
or not isinstance(priority, int) or isinstance(priority, bool)
|
|
or not isinstance(mandatory, bool)
|
|
or section_id in sections
|
|
):
|
|
raise ValueError("section metadata is invalid")
|
|
path = (self.root / filename).resolve()
|
|
if root not in path.parents or not path.is_file():
|
|
raise ValueError("section file is unavailable")
|
|
content = path.read_text(encoding="utf-8").strip()
|
|
if not content:
|
|
raise ValueError("section content is empty")
|
|
sections[section_id] = _Section(section_id, title, priority, mandatory, content)
|
|
|
|
def identifiers(value: Any, field: str) -> tuple[str, ...]:
|
|
if not isinstance(value, list) or not value or not all(isinstance(item, str) and item in sections for item in value):
|
|
raise ValueError(f"{field} is invalid")
|
|
return tuple(dict.fromkeys(value))
|
|
|
|
raw_phases = raw.get("phase_sections")
|
|
if not isinstance(raw_phases, dict) or "DEFAULT" not in raw_phases:
|
|
raise ValueError("phase sections are invalid")
|
|
phase_sections = {
|
|
phase: identifiers(section_ids, f"phase {phase}")
|
|
for phase, section_ids in raw_phases.items()
|
|
if isinstance(phase, str)
|
|
}
|
|
if len(phase_sections) != len(raw_phases):
|
|
raise ValueError("phase name is invalid")
|
|
operation_sections = {
|
|
atomic_id: identifiers(section_ids, f"operation {atomic_id}")
|
|
for atomic_id, section_ids in (raw.get("operation_sections") or {}).items()
|
|
if isinstance(atomic_id, str)
|
|
}
|
|
if not operation_sections or len(operation_sections) != len(raw.get("operation_sections") or {}):
|
|
raise ValueError("operation sections are invalid")
|
|
self._corpus = _Corpus(
|
|
version=version,
|
|
sections=sections,
|
|
phase_sections=phase_sections,
|
|
repair_sections=identifiers(raw.get("repair_sections"), "repair sections"),
|
|
final_sections=identifiers(raw.get("final_sections"), "final sections"),
|
|
operation_sections=operation_sections,
|
|
)
|
|
return self._corpus
|
|
except (OSError, ValueError, TypeError, json.JSONDecodeError) as error:
|
|
self._load_error = f"guidance_load_failed:{type(error).__name__}"
|
|
return None
|