109 lines
5.8 KiB
Python
109 lines
5.8 KiB
Python
"""Compact implementation profiles with exact aliases and source-grounded evidence."""
|
|
from __future__ import annotations
|
|
|
|
from suite_contract import SYSTEM, Problem, encoded
|
|
from suite_policy import evidence
|
|
|
|
PROFILE_FIELDS = {"target": 64, "work": 192, "setup": 96, "stimulus": 96,
|
|
"observations": 128, "machinery": 96, "variations": 96, "uncertainty": 96}
|
|
PROFILE_BYTES = 1152
|
|
BATCH_CASES = 24
|
|
BATCH_BYTES = 48 * 1024
|
|
MAX_PROFILE_BATCHES = 32
|
|
PROFILE_INSTRUCTIONS = (
|
|
"Extract a compact implementation profile for EVERY alias. These processing batches "
|
|
"are NOT families. Work must capture primary success-criteria implications; setup "
|
|
"preserves material preconditions; stimulus captures controls/actions; observations "
|
|
"captures measurement/assertions; machinery identifies special test mechanisms; "
|
|
"variations and uncertainty preserve qualitative distinctions and missing details. "
|
|
"Use null for unknowns. State only supplied facts, no invented implementation. "
|
|
"Keep profiles concise, under 1152 UTF-8 bytes each including evidence. Include one "
|
|
"short exact evidence quote from success_criteria if present, otherwise description. "
|
|
"Do not drop distinctions to force a merge. Output profiles keyed by exact alias."
|
|
)
|
|
|
|
|
|
def batches(source):
|
|
"""Pack complete cases deterministically; never split or truncate one case."""
|
|
result, current = [], []
|
|
for case in sorted(source["cases"], key=lambda c: c["alias"]):
|
|
if current and (len(current) >= BATCH_CASES or len(encoded(current + [case])) > BATCH_BYTES):
|
|
result.append({**source, "cases": current})
|
|
current = []
|
|
current.append(case)
|
|
if current:
|
|
result.append({**source, "cases": current})
|
|
if len(result) > MAX_PROFILE_BATCHES:
|
|
raise Problem("profile_batch_capacity", 422, batch_count=len(result))
|
|
return result
|
|
|
|
|
|
def profile_call(source, index):
|
|
"""Supply every generalized source field and require one result per alias."""
|
|
item = {"type": "object", "additionalProperties": False,
|
|
"required": list(PROFILE_FIELDS) + ["evidence"], "properties": {
|
|
k: {"type": ["string", "null"], "maxLength": size} for k, size in PROFILE_FIELDS.items()}}
|
|
item["properties"]["evidence"] = {"type": "object", "additionalProperties": False,
|
|
"required": ["field", "quote"], "properties": {
|
|
"field": {"enum": ["success_criteria", "description"]},
|
|
"quote": {"type": "string", "minLength": 1, "maxLength": 96}}}
|
|
aliases = sorted(c["alias"] for c in source["cases"])
|
|
return {"stage": "implementation_profiles", "batch_index": index,
|
|
"input": encoded({k: source[k] for k in ("campaign", "suite", "cases")}).decode(),
|
|
"system": SYSTEM + "\n" + PROFILE_INSTRUCTIONS,
|
|
"schema": {"type": "object", "additionalProperties": False, "required": ["profiles"],
|
|
"properties": {"profiles": {"type": "object", "additionalProperties": False,
|
|
"required": aliases, "properties": {a: item for a in aliases}}}}}
|
|
|
|
|
|
def validate_profiles(value, source):
|
|
"""Check profile coverage, size, nullable fields and exact source evidence."""
|
|
by_alias = {c["alias"]: c for c in source["cases"]}
|
|
if not isinstance(value, dict) or set(value) != {"profiles"} or not isinstance(value["profiles"], dict):
|
|
raise Problem("invalid_profile_output", 502)
|
|
if set(value["profiles"]) != set(by_alias):
|
|
raise Problem("invalid_case_assignments", 502, failure_stage="profile_aliases")
|
|
for alias, profile in value["profiles"].items():
|
|
if not isinstance(profile, dict) or set(profile) != set(PROFILE_FIELDS) | {"evidence"}:
|
|
raise Problem("invalid_profile_output", 502)
|
|
for key, limit in PROFILE_FIELDS.items():
|
|
text = profile[key]
|
|
if text is not None and (type(text) is not str or len(text) > limit):
|
|
raise Problem("invalid_profile_output", 502)
|
|
item = profile["evidence"]
|
|
if not isinstance(item, dict) or set(item) != {"field", "quote"}:
|
|
raise Problem("invalid_profile_output", 502)
|
|
expected = "success_criteria" if by_alias[alias].get("success_criteria") else "description"
|
|
if item["field"] != expected:
|
|
raise Problem("invalid_review_evidence", 502)
|
|
evidence([{**item, "alias": alias}], {alias}, by_alias)
|
|
if len(encoded(profile)) > PROFILE_BYTES:
|
|
raise Problem("profile_capacity", 422)
|
|
return value
|
|
|
|
|
|
def profile_source(source, profiles):
|
|
"""Use compact fields for global comparison; originals are reopened afterwards."""
|
|
cases = []
|
|
for alias in sorted(profiles):
|
|
p = profiles[alias]
|
|
def join(keys):
|
|
return "; ".join(k + ": " + (p[k] or "unknown") for k in keys)
|
|
cases.append({"alias": alias, "description": join(("target", "stimulus")),
|
|
"success_criteria": join(("work", "observations", "machinery")),
|
|
"preconditions": join(("setup",)),
|
|
"operating_condition": join(("variations", "uncertainty"))})
|
|
if any(len(encoded(c)) > 1280 for c in cases):
|
|
raise Problem("profile_capacity", 422)
|
|
return {**source, "cases": cases, "_profile_source": True}
|
|
|
|
|
|
def size_metadata(source):
|
|
"""Retain byte distributions only, allowing future representative fault tests."""
|
|
result = {"source_bytes": len(encoded({k: source[k] for k in ("campaign", "suite", "cases")}))}
|
|
for field in ("description", "preconditions", "case_type", "success_criteria"):
|
|
values = sorted(len((c.get(field) or "").encode()) for c in source["cases"])
|
|
result[field] = {"min": min(values), "max": max(values), "median": values[len(values)//2],
|
|
"p95": values[int((len(values)-1)*.95)]}
|
|
return result
|