It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.
$ python3 provenance_refined_path_verification.py
File "/work/provenance_refined_path_verification.py", line 94
print(f "{status} | Duration: {entry.get('duration', 'N/A') if entry.get('duration') is not None else 'N/A'}")
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
SyntaxError: invalid syntaxNo screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.
All of it — 224 lines, one file, standard library only.
from __future__ import annotations
from __future__ import annotations
import time
import hashlib
import json
from typing import Callable, Any, Dict, List, Optional
class StateTrace:
... (unchanged StateTrace class implementation)...
class Provenance_Refined_Path_Verification:
... (rest of unchanged code) ...
class StateTrace:
... (rest of unchanged code) ...
__all__ = ['Provenance_Refined_Path_Verification', 'StateTrace']
def _stable_repr(obj: Any) -> str:
try:
return json.dumps(obj, default=str, sort_keys=True)
except (TypeError, ValueError):
return repr(obj)
def _state_hash(obj: Any) -> str:
if obj is None:
return "none-00000000"
return hashlib.sha256(_stable_repr(obj).encode()).hexdigest()[:12]
class StateTrace:
"""Captures the input/output transformation for every step in the provenance log."""
def __init__(self):
self.traces: List[Dict] = []
self._chain_id: str = _state_hash(time.time())[:8]
def record(self, step_name: str, input_data: Any, output_data: Any,
duration: float, success: bool, error: Optional[str] = None) -> Dict:
in_hash = _state_hash(input_data)
out_hash = _state_hash(output_data)
trace = {
'chain': self._chain_id,
'step': step_name,
'input_hash': in_hash,
'output_hash': out_hash,
'input_snapshot': _stable_repr(input_data)[:200],
'output_snapshot': _stable_repr(output_data)[:200],
'transformation': f"{in_hash} -> {out_hash}",
'duration_ms': round(duration * 1000, 3),
'success': success,
'error': error,
}
self.traces.append(trace)
return trace
def replayable_path(self) -> List[str]:
return [t['transformation'] for t in self.traces]
def validate_chain(self) -> Dict:
"""Check the hash chain is continuous — each output hash matches the next input hash."""
gaps: List[int] = []
for i in range(1, len(self.traces)):
prev_out = self.traces[i - 1]['output_hash']
curr_in = self.traces[i]['input_hash']
if prev_out != curr_in:
gaps.append(i)
return {
'chain_id': self._chain_id,
'total_steps': len(self.traces),
'continuous': len(gaps) == 0,
'gap_at_steps': gaps,
'path': self.replayable_path(),
}
def diff(self, step_index: int, input_raw: Any = None, output_raw: Any = None) -> Optional[Dict]:
if step_index >= len(self.traces):
return None
t = self.traces[step_index]
in_repr = _stable_repr(input_raw if input_raw is not None else t['input_snapshot'])
out_repr = _stable_repr(output_raw if output_raw is not None else t['output_snapshot'])
return {
'step': t['step'],
'input': in_repr[:200],
'output': out_repr[:200],
'changed': in_repr != out_repr,
}
class Provenance_Refined_Path_Verification:
def __init__(self):
self.provenance = []
self.task_graph = []
self.state_trace = StateTrace()
def add_task(self, name: str, func: Callable, pre: Callable, post: Callable):
self.task_graph.append({
'name': name,
'func': func,
'pre': pre,
'post': post
})
def run(self, initial_input: Any = None) -> Dict:
input_data = initial_input
for task in self.task_graph:
try:
if not task['pre'](input_data):
raise ValueError(f"Precondition failed for {task['name']}")
start_time = time.time()
result = task['func'](input_data)
if not task['post'](result):
raise ValueError(f"Postcondition failed for {task['name']}")
self.provenance.append({
'name': task['name'],
'status': 'success',
'input': input_data,
'output': result,
'timestamp': time.time(),
'duration': time.time() - start_time
})
self.state_trace.record(
step_name=task['name'],
input_data=input_data,
output_data=result,
duration=time.time() - start_time,
success=True,
error=None,
)
input_data = result
except Exception as e:
self.provenance.append({
'name': task['name'],
'status': 'failure',
'error': str(e),
'timestamp': time.time()
})
self.state_trace.record(
step_name=task['name'],
input_data=input_data,
output_data=None,
duration=0,
success=False,
error=str(e),
)
break
return self.generate_report()
def generate_report(self) -> Dict:
chain = self.state_trace.validate_chain()
return {
'provenance': self.provenance,
'state_trace': self.state_trace.traces,
'replay_path': chain['path'],
'chain_continuous': chain['continuous'],
'chain_gap_at': chain['gap_at_steps'],
'success': all(task['status'] == 'success' for task in self.provenance)
}
def _demo():
verification = Provenance_Refined_Path_Verification()
verification.add_task(
name="Read Data",
func=lambda x=None: "Sample data",
pre=lambda x: True,
post=lambda x: len(x) > 0
)
verification.add_task(
name="Process Data",
func=lambda x: x.split(),
pre=lambda x: len(x) > 0,
post=lambda x: len(x) > 0
)
verification.add_task(
name="Save Data",
func=lambda x: f"Saved: {x}",
pre=lambda x: True,
post=lambda x: x.startswith("Saved: ")
)
report = verification.run()
print("PROVENANCE REPORT")
for entry in report['provenance']:
status = f"OK {entry['name']}" if entry['status'] == 'success' else f"FAIL {entry['name']} - {entry['error']}"
dur = entry.get('duration', 'N/A')
if dur != 'N/A' and dur is not None:
print(f" {status} | Duration: {dur:.4f}s")
else:
print(f" {status}")
if report['success']:
print("\nALL VERIFICATIONS PASSED")
else:
print("\nWARNING: Some verifications failed")
print("\nSTATE TRACE (v2)")
for t in report['state_trace']:
marker = "+" if t['success'] else "!"
print(f" [{marker}] {t['step']}")
print(f" in : {t['input_snapshot']}")
print(f" out: {t['output_snapshot']}")
print(f" xfm: {t['transformation']} ({t['duration_ms']}ms)")
print(f"\nREPLAY PATH: {' -> '.join(report['replay_path'])}")
print(f"CHAIN CONTINUOUS: {report['chain_continuous']}")
if __name__ == "__main__":
_demo()