From a90d251f39c31e64f324b8ffe66274fffd222101 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Wed, 28 Oct 2026 06:06:52 -0700 Subject: [PATCH] refactor(inspector): share the inspector's resolve as inspection_resolution inspect_prompt_evidence now calls inspection_resolution, which returns the resolution to use, the anchored result and whether the recorded saver chose it. Ingestion can call the same helper for files the default profile cannot decide (#247). --- tests/test_recorded_saver_anchor.py | 60 +++++++++ utils/prompt_inspector.py | 188 ++++++++++++++++++---------- 2 files changed, 183 insertions(+), 65 deletions(-) diff --git a/tests/test_recorded_saver_anchor.py b/tests/test_recorded_saver_anchor.py index 3ca852c6..580f2c45 100644 --- a/tests/test_recorded_saver_anchor.py +++ b/tests/test_recorded_saver_anchor.py @@ -375,5 +375,65 @@ class TestRecordedTexts(unittest.TestCase): self.assertEqual(evidence["resolution"]["sampler_id"], sampler) +class TestInspectionResolution(unittest.TestCase): + """``inspection_resolution``: the inspector's resolve and recorded anchor, + shared with ingestion (#247), so the two cannot drift.""" + + def _result(self, graph, saver=None, *, as_text=True, name="IMG_GEN_00371_.png"): + from utils.prompt_inspector import inspection_resolution + + prompt = json.dumps(graph) if as_text else graph + return inspection_resolution({"prompt": prompt}, name, saver) + + def test_the_recorded_saver_names_its_branch_by_record(self): + result = self._result(shapes.shared_prefix(), _saver(2)) + self.assertTrue(result.by_record) + self.assertFalse(result.malformed) + self.assertEqual(result.resolution["state"], "resolved") + self.assertEqual(_positive(result.resolution), shapes.BRANCH_TEXTS[1]) + self.assertIs(result.anchored, result.resolution) + + def test_without_a_saver_nothing_is_anchored(self): + result = self._result(shapes.shared_prefix()) + self.assertEqual(result.resolution["state"], "ambiguous") + self.assertIsNone(result.anchored) + self.assertFalse(result.by_record) + + def test_a_saver_that_names_nothing_keeps_the_open_result(self): + result = self._result(shapes.shared_prefix(), "missing") + self.assertEqual(result.resolution["state"], "ambiguous") + self.assertIsNotNone(result.anchored) + self.assertFalse(result.by_record) + + def test_a_file_that_resolves_alone_is_never_anchored(self): + result = self._result( + shapes.branch_bench(), _saver(4), name="s45_branch_2_00001_.png" + ) + self.assertEqual(_positive(result.resolution), shapes.BRANCH_TEXTS[1]) + self.assertIsNone(result.anchored) + self.assertFalse(result.by_record) + + def test_parsed_metadata_resolves_as_the_file_text_does(self): + # The monitor hands parsed dicts; the scan hands the file's strings. + parsed = self._result(shapes.shared_prefix(), _saver(3), as_text=False) + text = self._result(shapes.shared_prefix(), _saver(3)) + self.assertEqual(parsed.resolution, text.resolution) + + def test_malformed_metadata_is_reported_as_malformed(self): + from utils.prompt_inspector import inspection_resolution + + result = inspection_resolution({"prompt": "{not json"}, None, None) + self.assertTrue(result.malformed) + self.assertEqual(result.resolution["state"], "unresolved") + self.assertEqual(result.resolution["reason_code"], "malformed") + + def test_metadata_without_a_role_record_is_absent(self): + from utils.prompt_inspector import inspection_resolution + + result = inspection_resolution({"workflow": "{}"}, None, "9") + self.assertFalse(result.malformed) + self.assertEqual(result.resolution["state"], "absent") + + if __name__ == "__main__": unittest.main() diff --git a/utils/prompt_inspector.py b/utils/prompt_inspector.py index 1daaef6c..05a0b76b 100644 --- a/utils/prompt_inspector.py +++ b/utils/prompt_inspector.py @@ -1,4 +1,10 @@ -"""Read prompt evidence without inference, indexing, or decoding image pixels.""" +"""Read prompt evidence without inference, indexing, or decoding image pixels. + +``inspection_resolution`` is the inspector's resolve (inspection profile, then +the recorded saver anchor, #386). Ingestion (the output scan, through +``utils.file_prompt``) calls it too, only for files the default profile +cannot decide (#247), so the inspector and ingestion cannot drift. +""" import hashlib import io @@ -571,6 +577,93 @@ def _recorded_execution(resolved: dict, texts: Mapping) -> dict | None: return {"prompts": prompts, "resolution": resolution} +_ABSENT_RESOLUTION = { + "state": "absent", + "sampler_id": None, + "reason": "No parsed prompt metadata", + "reason_code": None, + "node": None, +} +_MALFORMED_RESOLUTION = { + "state": "unresolved", + "sampler_id": None, + "reason": MALFORMED_EVIDENCE_REASON, + "reason_code": "malformed", + "node": None, +} + + +@dataclass(frozen=True) +class InspectionResult: + """The inspector's resolution of one file's metadata (#386, #247). + + ``resolution`` is the result to use. ``anchored`` is the result anchored + on the recorded saver, kept or not (None when no anchored resolve ran). + ``by_record`` says the anchored result was kept: history's saver chose + the samplers and the result names the positive. ``malformed`` marks + metadata the inspector cannot read as a graph. + """ + + resolution: dict + anchored: dict | None = None + by_record: bool = False + malformed: bool = False + + +def inspection_resolution( + metadata: Mapping, + file_name: str | None, + recorded_saver_id: str | None, + *, + profile: ResolverProfile = INSPECTION_PROFILE, +) -> InspectionResult: + """Resolve raw file metadata as the inspector does. + + Keeps only the role records, parses a string ``prompt`` and attaches the + parsed ``workflow`` only when the graph may hold a PromptManager node. + Resolves with ``profile``; when that leaves the prompt open and + ``recorded_saver_id`` is given, resolves again anchored on that saver + and keeps the anchored result only when it names the positive. + """ + workflow = metadata.get("workflow") + canonical = { + key: value for key, value in metadata.items() if key in _ROLE_METADATA_KEYS + } + if not canonical: + return InspectionResult(dict(_ABSENT_RESOLUTION)) + malformed = InspectionResult(dict(_MALFORMED_RESOLUTION), malformed=True) + if "prompt" not in canonical and "parameters" not in canonical: + return malformed + if isinstance(canonical.get("prompt"), str): + try: + canonical["prompt"] = json.loads(canonical["prompt"]) + except (ValueError, RecursionError): + return malformed + workflow = ( + _parsed_workflow(workflow) if _may_hold_prompt_manager(canonical) else None + ) + if workflow is not None and "prompt" in canonical: + # Only to recognize overwritten PromptManager text (R1); never a role. + canonical["workflow"] = workflow + resolved = resolve_prompt_pair(canonical, profile=profile, file_name=file_name) + if resolved["state"] not in _UNDECIDED or recorded_saver_id is None: + return InspectionResult(resolved) + anchored = resolve_prompt_pair( + canonical, + profile=profile, + file_name=file_name, + recorded_saver_id=recorded_saver_id, + ) + # Only a result whose samplers history's saver chose is about this file. + if ( + anchored.get("recorded_anchor") is True + and anchored["state"] not in _UNDECIDED + and isinstance(anchored["positive"], dict) + ): + return InspectionResult(anchored, anchored, True) + return InspectionResult(resolved, anchored) + + def inspect_prompt_evidence( metadata: dict, saved: dict | None = None, @@ -593,22 +686,13 @@ def inspect_prompt_evidence( ``execution``. Inspection routes resolve with ``INSPECTION_PROFILE``: bounded work, not - graph size. Ingestion (scan, live monitor, admin import) calls the resolver - directly with its default profile. ``resolution`` carries the closed - ``reason_code`` and, for a blocking node, ``node: {id, class_type}``. + graph size. The resolve itself is ``inspection_resolution``, which + ingestion also calls for files its default profile cannot decide (#247). + ``resolution`` carries the closed ``reason_code`` and, for a blocking + node, ``node: {id, class_type}``. """ - workflow = metadata.get("workflow") - metadata = { - key: value for key, value in metadata.items() if key in _ROLE_METADATA_KEYS - } - resolution = { - "state": "unresolved", - "sampler_id": None, - "reason": MALFORMED_EVIDENCE_REASON, - "reason_code": "malformed", - "node": None, - } - if not metadata: + roles = {key for key in metadata if key in _ROLE_METADATA_KEYS} + if not roles: result = {role: _side("absent") for role in ("positive", "negative")} if saved: context = saved.get("pair_context") @@ -621,56 +705,30 @@ def inspect_prompt_evidence( "verified against this file." ), } - return { - "prompts": result, - "resolution": { - "state": "absent", - "sampler_id": None, - "reason": "No parsed prompt metadata", - "reason_code": None, - "node": None, - }, - } - canonical = dict(metadata) - malformed = { - "prompts": _explained( - {role: _side("unresolved") for role in ("positive", "negative")}, - resolution, - ), - "resolution": resolution, - } - if "prompt" not in canonical and "parameters" not in canonical: - return malformed - if isinstance(canonical.get("prompt"), str): - try: - canonical["prompt"] = json.loads(canonical["prompt"]) - except (ValueError, RecursionError): - return malformed - workflow = ( - _parsed_workflow(workflow) if _may_hold_prompt_manager(canonical) else None + return {"prompts": result, "resolution": dict(_ABSENT_RESOLUTION)} + inspected = inspection_resolution( + metadata, file_name, recorded_saver_id, profile=profile ) - if workflow is not None and "prompt" in canonical: - # Only to recognize overwritten PromptManager text (R1); never a role. - canonical["workflow"] = workflow - resolved = resolve_prompt_pair(canonical, profile=profile, file_name=file_name) - recorded = False - if resolved["state"] in _UNDECIDED and recorded_saver_id is not None: - anchored = resolve_prompt_pair( - canonical, - profile=profile, - file_name=file_name, - recorded_saver_id=recorded_saver_id, - ) - # Only a result whose samplers history's saver chose is about this file. - if anchored.get("recorded_anchor") is True: - if anchored["state"] not in _UNDECIDED and isinstance( - anchored["positive"], dict - ): - resolved, recorded = anchored, True - elif recorded_texts: - executed = _recorded_execution(anchored, recorded_texts) - if executed is not None: - return executed + if inspected.malformed: + resolution = inspected.resolution + return { + "prompts": _explained( + {role: _side("unresolved") for role in ("positive", "negative")}, + resolution, + ), + "resolution": resolution, + } + resolved, recorded = inspected.resolution, inspected.by_record + anchored = inspected.anchored + if ( + not recorded + and recorded_texts + and anchored is not None + and anchored.get("recorded_anchor") is True + ): + executed = _recorded_execution(anchored, recorded_texts) + if executed is not None: + return executed state = resolved["state"] result = {} for role in ("positive", "negative"):