@@ -1,5 +1,6 @@
 """Generated save privacy; no generated source inspection or synthetic traces."""
 import asyncio
+import json
 import os
 import uuid
 
@@ -9,7 +10,84 @@
 from teaql.provider.sqlite import create_sqlite_service
 from teaql.runtime import UserContext
 
-from main import AuditSink, ObservedService, new, reset
+from main import AuditSink, ObservedService, new, reset, readback_fact, scalar
+
+
+def complete_lineage(nodes, root_id, reason, boundary):
+    """Empty/redacted-away responsibility is not a successful privacy projection."""
+    assert len(nodes) == 1, boundary + ': complete typed root lineage required'
+    node = nodes[0]
+    assert (node.kind, node.name, node.entity_id, node.comment) == (
+        'auditReason', 'CustomerOrder', root_id, reason), boundary + ': wrong root responsibility'
+
+
+def assert_private_lineage(service, sink, context, root_id, reason, identities, secrets,
+                           expected_bindings, phase):
+    safe_reason = reason
+    for secret in secrets:
+        safe_reason = safe_reason.replace(secret, '[REDACTED]')
+    expected = set(identities)
+    assert len(expected) == len(identities), 'privacy graph needs distinct typed identities'
+    count = len(identities)
+    assert len(service.raw_requests) == len(service.results) == len(sink.events) == count
+    assert service.finishes == ['commit'], 'privacy graph must commit before audit delivery'
+    assert len(context.sql_logs()) == count * 2
+    commands = []
+    seen = set()
+    for request, result in zip(service.raw_requests, service.results):
+        command = request._data
+        identity = (command.entity, scalar(getattr(command, 'id', None) or command.values.get('id')))
+        assert identity in expected and identity not in seen, 'actual privacy command identity mismatch'
+        seen.add(identity)
+        assert request.comment() == result.comment == reason, 'privacy must not rewrite command reason'
+        complete_lineage(request.mutation_lineage, root_id, reason, 'raw privacy command')
+        complete_lineage(result.mutation_lineage, root_id, reason, 'raw provider mutation')
+        assert len(result.statements) == 2 and result.statements[0].affected_rows == 1
+        for field, value in expected_bindings.get(command.entity, {}).items():
+            assert scalar(command.values[field]) == value, 'privacy must not rewrite command bindings'
+            assert value in [scalar(parameter) for parameter in result.statements[0].parameters], 'privacy must not rewrite provider parameters'
+        commands.append({'entity': identity[0], 'id': identity[1], 'comment': request.comment(),
+                         'lineage': readback_nodes(request.mutation_lineage)})
+    assert seen == expected
+    for index, (request, result) in enumerate(zip(service.raw_requests, service.results)):
+        write, read = context.sql_logs()[index * 2:index * 2 + 2]
+        for fact in (write, read):
+            complete_lineage(fact.mutation_lineage, root_id, safe_reason, 'safe privacy SQL')
+            assert fact.comment == fact.audit_reason == safe_reason, 'safe SQL must retain public reason'
+            assert fact.execution_outcome == 'success'
+            assert [node.kind for node in fact.trace_path] == [
+                'operation', 'entity' if fact is write else 'request', 'provider', 'sql']
+            assert fact.trace_path[0].name == 'CustomerOrder'
+            assert fact.trace_path[2].name == 'sqlite'
+        assert write.trace_path[1].name == request._data.entity
+        assert write.trace_path[-1].name == service.actions[index]
+        assert read.trace_path[-1].name == 'select'
+        assert write.affected_rows == read.result_count == 1 and read.affected_rows is None
+        assert read.purpose == 'verify the persisted mutation result'
+        assert write.ended_at <= read.started_at
+        assert not result.statements[0].statements and not result.statements[1].statements
+    audits = []
+    seen = set()
+    for event in sink.events:
+        identity = (event.entity, scalar(event.entity_id))
+        assert identity in expected and identity not in seen, 'committed privacy audit identity mismatch'
+        seen.add(identity)
+        complete_lineage(event.trace_chain, root_id, safe_reason, 'safe privacy audit')
+        audits.append({'entity': identity[0], 'id': identity[1],
+                       'lineage': readback_nodes(event.trace_chain)})
+    assert seen == expected
+    for secret in secrets:
+        assert secret not in repr(context.sql_logs()) and secret not in repr(sink.events)
+    print('PRIVATE_LINEAGE_OBSERVED ' + json.dumps({
+        'phase': phase, 'rootId': root_id, 'rawReason': reason, 'safeReason': safe_reason,
+        'commands': commands, 'sql': [readback_fact(fact) for fact in context.sql_logs()],
+        'audit': audits}, sort_keys=True))
+    print('PASS Python complete private lineage: raw commands, safe SQL/readback and committed audit')
+
+
+def readback_nodes(nodes):
+    return [{'Kind': node.kind, 'Name': node.name, 'EntityId': node.entity_id,
+             'Comment': node.comment} for node in nodes]
 
 
 async def main():
@@ -52,6 +130,10 @@ async def main():
         for metadata in service.results:
             assert len(metadata.statements) == 2
             assert metadata.comment == reason  # raw owned intent is not rewritten
+        assert_private_lineage(service, sink, context, E.customer_order(root).id().eval(), reason,
+            [('CustomerOrder', E.customer_order(root).id().eval()),
+             ('OrderItem', E.order_item(item).id().eval()), ('Payment', E.payment(payment).id().eval())],
+            [secret, old], {'Payment': {'reference_code': secret}}, phase)
         loaded = await Q.payments().with_id_is(E.payment(payment).id().eval()).limit(1).comment(
             'reload authoritative payment').purpose('verify safe logs do not alter business data').execute_for_one(context)
         assert E.payment(loaded).reference_code().eval() == secret
@@ -77,6 +159,9 @@ async def main():
     assert secret not in repr(context.sql_logs()), 'loaded delete value leaked into SQL intent'
     assert secret not in repr(sink.events), 'loaded delete value leaked into committed audit'
     assert all(metadata.comment == reason for metadata in service.results)
+    assert_private_lineage(service, sink, context, E.customer_order(root).id().eval(), reason,
+        [('CustomerOrder', E.customer_order(root).id().eval()), ('Payment', payment_id)],
+        [secret], {}, 'delete')
     missing = await Q.payments().with_id_is(payment_id).limit(1).comment('verify deleted payment').purpose(
         'prove soft delete hides the row').execute_for_one(context)
     assert missing is None
@@ -94,10 +179,14 @@ async def main():
         'load all order fields').purpose('retain an unchanged private description').execute_for_one(context)
     root.update_order_number(label + '-revised')
     reset(service, sink, context)
-    await root.audit_as(f'renumber {private_description}').save(context)
+    reason = f'renumber {private_description}'
+    await root.audit_as(reason).save(context)
     assert len(service.requests) == len(sink.events) == 1
     assert private_description not in repr(context.sql_logs()), 'unchanged loaded scalar leaked into SQL intent'
     assert private_description not in repr(sink.events), 'unchanged loaded scalar leaked into audit'
+    assert_private_lineage(service, sink, context, E.customer_order(root).id().eval(), reason,
+        [('CustomerOrder', E.customer_order(root).id().eval())], [private_description],
+        {'CustomerOrder': {'order_number': label + '-revised'}}, 'unchanged')
     reloaded = await Q.customer_orders().with_id_is(E.customer_order(root).id().eval()).limit(1).comment(
         'verify unchanged description').purpose('prove provenance did not alter business values').execute_for_one(context)
     assert E.customer_order(reloaded).description().eval() == private_description
 
@@ -81,6 +81,8 @@ for attempt in first second; do
   python "$example/mutation_privacy.py" | tee "$privacy_log"
   rg -Fq 'PASS: Python generated loaded delete privacy; 2 writes/2 reads/2 audits; independent next request' "$privacy_log"
   rg -Fq 'PASS: Python generated unchanged scalar privacy; 1 write/1 read/1 audit' "$privacy_log"
+  [[ "$(rg -c '^PRIVATE_LINEAGE_OBSERVED ' "$privacy_log")" == 4 ]]
+  [[ "$(rg -c '^PASS Python complete private lineage:' "$privacy_log")" == 4 ]]
   echo "Assigned identities $attempt on $TEAQL_TRACE_CHAIN_SHARED_DB without cleanup"
   assigned_log="$(mktemp -t teaql-python-assigned.XXXXXX.log)"
   TEAQL_TRACE_CHAIN_DB="$TEAQL_TRACE_CHAIN_SHARED_DB" \