Evidence profile: opt-in identity of the exact graph input

tests/test_evidence_profile.py

@@ -360,5 +360,161 @@ def test_opt_in_invalid_graph_still_fails_closed(self):
         self.assertNotIn(b'unrecognized arguments', process.stderr)
 
 
+class GraphInputIdentity(unittest.TestCase):
+    """The optional digest identifies the single consumed byte buffer, not a canonicalized graph."""
+
+    MEANING = ('identity of graph input bytes only; not freshness, Git provenance, '
+               'program identity or acceptance')
+
+    def invoke(self, raw, *args):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = pathlib.Path(tmp) / 'private-local-name.json'
+            path.write_bytes(raw)
+            flags = ['-O'] if sys.flags.optimize else []
+            result = subprocess.run(
+                [sys.executable, '-B', *flags, '-S', str(ROOT / 'tools' / 'evidence_profile.py'),
+                 '--graph', str(path), *args], capture_output=True, timeout=60)
+            self.assertEqual(path.read_bytes(), raw)
+            self.assertNotIn(str(path).encode(), result.stdout)
+            return result
+
+    def expected_identity(self, raw):
+        return {'bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest(),
+                'meaning': self.MEANING}
+
+    def test_json_binds_exact_bytes_and_composes_with_eligibility(self):
+        graph = synthetic()
+        graph['object'] = 'INPUT-IDENTITY-\u03c0'
+        texts = [json.dumps(graph, ensure_ascii=False),
+                 json.dumps(graph, ensure_ascii=False, indent=2),
+                 json.dumps(graph, ensure_ascii=False, indent=2).replace('\n', '\r\n') + '\r\n',
+                 json.dumps(graph, ensure_ascii=True, sort_keys=True)]
+        digests = set()
+        for text in texts:
+            raw = text.encode('utf-8')
+            with self.subTest(raw_bytes=len(raw)):
+                args = ('--json', '--eligibility', '--node', 'math.top', '--node', 'math.mid')
+                plain = self.invoke(raw, *args)
+                identified = self.invoke(raw, '--input-identity', *args)
+                self.assertEqual((plain.returncode, plain.stderr), (0, b''))
+                self.assertEqual((identified.returncode, identified.stderr), (0, b''))
+                data = json.loads(identified.stdout)
+                self.assertEqual(data.pop('input_identity'), self.expected_identity(raw))
+                self.assertEqual(data, json.loads(plain.stdout))
+                self.assertEqual([p['node'] for p in data['profiles']], ['math.top', 'math.mid'])
+                for profile in data['profiles']:
+                    self.assertEqual(profile['controlling_eligibility']['decision'],
+                                     GATE.promotion_allowed(graph, profile['node']))
+                digests.add(hashlib.sha256(raw).hexdigest())
+        self.assertEqual(len(digests), len(texts))
+
+    def test_text_adds_one_identity_without_disclosing_paths(self):
+        raw = json.dumps(synthetic()).encode('utf-8')
+        for extra in ((), ('--eligibility',)):
+            with self.subTest(extra=extra):
+                plain = self.invoke(raw, *extra)
+                identified = self.invoke(raw, '--input-identity', *extra)
+                self.assertEqual((plain.returncode, plain.stderr), (0, b''))
+                self.assertEqual((identified.returncode, identified.stderr), (0, b''))
+                prefix = ('Graph input SHA-256: %s (%d bytes)\n%s\n\n' % (
+                    hashlib.sha256(raw).hexdigest(), len(raw), self.MEANING)).encode('utf-8')
+                self.assertEqual(identified.stdout, prefix + plain.stdout)
+
+    def test_invalid_inputs_refuse_before_any_identity_or_report(self):
+        valid = json.dumps(synthetic())
+        bad_class = synthetic()
+        bad_class['nodes']['math.top']['classification'] = 'CLOSED'
+        cycle = synthetic()
+        cycle['edges'].append({'from': 'math.leaf', 'to': 'math.top',
+                               'required': True, 'relation': 'uses'})
+        cases = [(b'{"schema_version":1,' + valid[1:].encode(), b'duplicate JSON key'),
+                 (valid.replace('"SYNTHETIC"', 'NaN').encode(), b'non-finite JSON constant'),
+                 (valid.replace('"SYNTHETIC"', 'Infinity').encode(), b'non-finite JSON constant'),
+                 (json.dumps(bad_class).encode(), b'unknown node classification'),
+                 (json.dumps(cycle).encode(), b'required dependency cycle'),
+                 (b'{not-json', b'JSONDecodeError'),
+                 (b'\xff', b'UnicodeDecodeError')]
+        for raw, diagnostic in cases:
+            for output in ((), ('--json',)):
+                with self.subTest(diagnostic=diagnostic, output=output):
+                    result = self.invoke(raw, '--input-identity', *output)
+                    self.assertEqual((result.returncode, result.stdout), (1, b''))
+                    self.assertIn(diagnostic, result.stderr)
+                    self.assertNotIn(b'unrecognized arguments', result.stderr)
+
+    def test_unknown_node_and_missing_path_emit_no_partial_identity(self):
+        raw = json.dumps(synthetic()).encode()
+        for output in ((), ('--json',)):
+            result = self.invoke(raw, '--input-identity', '--node', 'math.mid',
+                                 '--node', 'math.absent', *output)
+            self.assertEqual((result.returncode, result.stdout), (1, b''))
+            self.assertIn(b'unknown node: math.absent', result.stderr)
+        with tempfile.TemporaryDirectory() as tmp:
+            missing = pathlib.Path(tmp) / 'absent.json'
+            out = io.StringIO()
+            with contextlib.redirect_stdout(out):
+                try:
+                    ep.main(['--graph', str(missing), '--input-identity', '--json'])
+                except FileNotFoundError:
+                    pass
+                except SystemExit as exc:
+                    self.fail('input identity CLI unavailable: %s' % exc)
+                else:
+                    self.fail('missing graph unexpectedly accepted')
+            self.assertEqual(out.getvalue(), '')
+
+    def test_identity_and_report_use_one_read_despite_backing_path_change(self):
+        from unittest import mock
+        first = synthetic()
+        first['object'] = 'FIRST-CONSUMED'
+        later = synthetic()
+        later['object'] = 'LATER-PATH-CONTENT'
+        raw = json.dumps(first).encode()
+        replacement = json.dumps(later).encode()
+        with tempfile.TemporaryDirectory() as tmp:
+            path = pathlib.Path(tmp) / 'graph.json'
+            path.write_bytes(raw)
+            original = pathlib.Path.read_bytes
+            calls = []
+
+            def capture_then_change(p):
+                data = original(p)
+                if p == path:
+                    calls.append(p)
+                    # The parser and hash must consume the returned bytes, not reread this path.
+                    path.write_bytes(replacement)
+                return data
+
+            out, err = io.StringIO(), io.StringIO()
+            with mock.patch.object(pathlib.Path, 'read_bytes', capture_then_change):
+                with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+                    try:
+                        code = ep.main(['--graph', str(path), '--input-identity', '--json'])
+                    except SystemExit as exc:
+                        self.fail('input identity CLI unavailable: %s; %s' % (exc, err.getvalue()))
+            self.assertEqual(code, 0)
+            self.assertEqual(err.getvalue(), '')
+            self.assertEqual(calls, [path])
+            data = json.loads(out.getvalue())
+            self.assertEqual(data['object'], 'FIRST-CONSUMED')
+            self.assertEqual(data['input_identity'], self.expected_identity(raw))
+            self.assertEqual(path.read_bytes(), replacement)
+
+    def test_no_flag_keeps_existing_loader_and_report_contract(self):
+        from unittest import mock
+        graph = synthetic()
+        raw = json.dumps(graph).encode()
+        with tempfile.TemporaryDirectory() as tmp:
+            path = pathlib.Path(tmp) / 'graph.json'
+            path.write_bytes(raw)
+            # No flag means no new binary acquisition path at all.
+            with mock.patch.object(pathlib.Path, 'read_bytes', side_effect=AssertionError('new read without opt-in')):
+                code, text = run(['--graph', str(path), '--json'])
+            self.assertEqual(code, 0)
+            self.assertEqual(json.loads(text), ep.report(GATE, graph, ep.roots(graph)))
+            self.assertNotIn('input_identity', json.loads(text))
+            self.assertEqual(path.read_bytes(), raw)
+
+
 if __name__ == '__main__':
     unittest.main()

tools/evidence_profile.py

@@ -17,9 +17,16 @@
 reasons. This describes the supplied graph, not review sufficiency, an executed promotion or theorem
 acceptance. The default text and JSON output are unchanged when this option is absent.
 
-Usage: python3 -B -S tools/evidence_profile.py [--graph PATH] [--node ID] [--json] [--eligibility]
+With --input-identity, include the SHA-256 and length of the exact graph bytes consumed. The
+single captured buffer is both parsed and hashed; different serializations have different identities.
+This is not freshness, authenticated Git provenance, program/runtime identity or acceptance evidence.
+No path is reported. Without the flag, the existing loader and output remain unchanged.
+
+Usage: python3 -B -S tools/evidence_profile.py [--graph PATH] [--node ID] [--json] [--eligibility] [--input-identity]
 """
 import argparse
+import hashlib
+import io
 import importlib.util
 import json
 import pathlib
@@ -207,22 +214,55 @@ def report(gate, graph, node_ids, include_eligibility=False):
     }
 
 
+
+def load_graph_with_identity(gate, path):
+    """Parse and identify one captured buffer using the gate's unchanged strict validation.
+
+    TextIOWrapper matches Path.read_text's default decoding and universal-newline behavior;
+    the fingerprint covers original bytes, not the decoded or reserialized JSON.
+    """
+    raw = path.read_bytes()
+    with io.TextIOWrapper(io.BytesIO(raw)) as stream:
+        graph = gate.load_json_strict(stream.read())
+    gate.validate_graph_fail_closed(graph)
+    identity = {
+        'bytes': len(raw),
+        'sha256': hashlib.sha256(raw).hexdigest(),
+        'meaning': 'identity of graph input bytes only; not freshness, Git provenance, '
+                   'program identity or acceptance',
+    }
+    return graph, identity
+
+
 def main(argv=None):
     ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
     ap.add_argument('--graph', default=str(GATE_DIR / 'GRAPH.json'))
     ap.add_argument('--node', action='append', help='node id (repeatable); default: every math.* root')
     ap.add_argument('--json', action='store_true', help='print the profiles as JSON instead of trees')
     ap.add_argument('--eligibility', action='store_true',
                     help='include the selected-node gate decision; not theorem acceptance')
+    ap.add_argument('--input-identity', action='store_true',
+                    help='include exact graph-input byte identity, not provenance or acceptance')
     args = ap.parse_args(argv)
     gate = load_gate()
-    graph = gate.load_graph(pathlib.Path(args.graph))
+    identity = None
+    if args.input_identity:
+        graph, identity = load_graph_with_identity(gate, pathlib.Path(args.graph))
+    else:
+        graph = gate.load_graph(pathlib.Path(args.graph))
     node_ids = args.node or roots(graph)
     for nid in node_ids:
         gate.require_node(graph, nid)
     if args.json:
-        print(json.dumps(report(gate, graph, node_ids, args.eligibility), indent=2, sort_keys=True))
+        data = report(gate, graph, node_ids, args.eligibility)
+        if identity is not None:
+            data['input_identity'] = identity
+        print(json.dumps(data, indent=2, sort_keys=True))
         return 0
+    if identity is not None:
+        print('Graph input SHA-256: %s (%d bytes)' % (identity['sha256'], identity['bytes']))
+        print(identity['meaning'])
+        print()
     for index, nid in enumerate(node_ids):
         if index:
             print()