I built an independent file journal to retain item state beyond one conversation or kernel. It executes no business operation: it saves intent, caller-declared completion and review-needed state. Reopening and injected write-failure tests show both what the record preserves and where its guarantees end.
Use four labels to establish the record scope
Project scope, program, job and item form one identity. Its digest determines the filename, and the retained document must contain that same identity. Labels are not concatenated into arbitrary child paths. The caller supplies an absolute directory it owns and trusts.
Declared identities are not authenticated account ownership. The example also supplies no concurrent multiprocess writer: an instance-local lock protects only its short metadata changes. It is a component for reviewing local persistence, not a multi-user server journal or an operating-system filesystem sandbox.
Persist intent before returning a claim
Begin validates input and definition fingerprints, saves a started document and only then returns a claim. It accepts no execution callback. An adapter must separately enforce that no external operation occurs before this acknowledgement; the journal cannot control arbitrary code outside its methods.
Saving writes a unique temporary file, flushes and synchronizes it, then replaces the destination. Failure grants no successful claim and attempts temporary cleanup. This is not a database transaction covering every power-loss or directory-synchronization condition. The explanation identifies the file-replacement boundary actually implemented.
Apply different reopening rules to completion and interruption
Matching succeeded records return the retained receipt for reuse. Started or review-needed records reject another begin. Reopening a kernel cannot silently convert unresolved work into a fresh execution. Mark-review exposes uncertainty; it does not promote a record to completion.
Completion compares the current claim and fingerprints before storing a short caller-supplied receipt. That receipt is still an assertion, not verified remote success. The module always reports that no work was executed here. There is no service-storage or publication observer, so retained declaration and independently verified business outcome remain distinct.
Explore the implementation
Complete independent synthetic implementation
"""Independent file-backed synthetic item journal; no external work executor."""
from pathlib import Path
from dataclasses import dataclass
from threading import RLock
from hashlib import sha256
from uuid import uuid4
import json,os,re
class RecordError(ValueError):pass
def _label(value):
if not isinstance(value,str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,79}',value):raise RecordError('INVALID_LABEL')
def _identity(value):
if type(value) not in (tuple,list) or len(value)!=4:raise RecordError('INVALID_IDENTITY')
for part in value:_label(part)
return tuple(value)
def _hash(value):
if not isinstance(value,str) or not re.fullmatch(r'[a-f0-9]{64}',value):raise RecordError('INVALID_HASH')
def input_digest(inputs):
if type(inputs) is not dict or len(inputs)>20:raise RecordError('INVALID_INPUTS')
for key,value in inputs.items():
_label(key)
if not isinstance(value,str) or len(value)>20000:raise RecordError('INVALID_INPUTS')
return sha256(json.dumps(inputs,ensure_ascii=False,sort_keys=True,separators=(',',':')).encode('utf-8')).hexdigest()
@dataclass(frozen=True)
class Claim:
identity:tuple
input_hash:str
definition_hash:str
claim_id:str
class JobRecords:
def __init__(self,root):
self.root=Path(root)
if not self.root.is_absolute():raise RecordError('ABSOLUTE_ROOT_REQUIRED')
self.root.mkdir(parents=True,exist_ok=True)
self._lock=RLock()
def _path(self,identity):
identity=_identity(identity)
name=sha256(json.dumps(identity,separators=(',',':')).encode('utf-8')).hexdigest()
return self.root/(name+'.json')
def _read(self,identity):
path=self._path(identity)
try:raw=json.loads(path.read_text('utf-8'))
except FileNotFoundError:return None
except (OSError,ValueError) as error:raise RecordError('RECORD_INVALID') from error
keys={'version','identity','input_hash','definition_hash','claim_id','status','receipt'}
if type(raw) is not dict or set(raw)!=keys or type(raw['version']) is not int or raw['version']!=1:raise RecordError('RECORD_INVALID')
if raw['identity']!=list(identity) or raw['status'] not in ['started','succeeded','needs_review']:raise RecordError('RECORD_INVALID')
try:_hash(raw['input_hash']);_hash(raw['definition_hash'])
except RecordError as error:raise RecordError('RECORD_INVALID') from error
if not isinstance(raw['claim_id'],str) or not re.fullmatch(r'[a-f0-9]{32}',raw['claim_id']):raise RecordError('RECORD_INVALID')
if raw['status']=='succeeded':
if not isinstance(raw['receipt'],str) or not raw['receipt'].strip() or len(raw['receipt'])>200:raise RecordError('RECORD_INVALID')
elif raw['receipt'] is not None:raise RecordError('RECORD_INVALID')
return raw
def _save(self,identity,record):
destination=self._path(identity)
temporary=self.root/(uuid4().hex+'.tmp')
try:
with temporary.open('x',encoding='utf-8',newline='\n') as stream:
json.dump(record,stream,ensure_ascii=False,sort_keys=True)
stream.flush();os.fsync(stream.fileno())
os.replace(temporary,destination)
finally:
if temporary.exists():temporary.unlink()
def begin(self,identity,inputs,definition_hash):
identity=_identity(identity);fingerprint=input_digest(inputs);_hash(definition_hash)
with self._lock:
record=self._read(identity)
if record:
if record['input_hash']!=fingerprint:raise RecordError('INPUT_CHANGED')
if record['definition_hash']!=definition_hash:raise RecordError('DEFINITION_CHANGED')
if record['status']=='succeeded':return {'decision':'reuse','receipt':record['receipt'],'executed_here':False}
raise RecordError('REVIEW_REQUIRED')
claim=Claim(identity,fingerprint,definition_hash,uuid4().hex)
self._save(identity,{'version':1,'identity':list(identity),'input_hash':fingerprint,'definition_hash':definition_hash,'claim_id':claim.claim_id,'status':'started','receipt':None})
return claim
def complete(self,claim,receipt):
if type(claim) is not Claim:raise RecordError('INVALID_CLAIM')
if not isinstance(receipt,str) or not receipt.strip() or len(receipt)>200:raise RecordError('INVALID_RECEIPT')
with self._lock:
record=self._read(claim.identity)
if not record or (record['claim_id'],record['input_hash'],record['definition_hash'])!=(claim.claim_id,claim.input_hash,claim.definition_hash):raise RecordError('CLAIM_MISMATCH')
if record['status']!='started':raise RecordError('NOT_STARTED')
self._save(claim.identity,{**record,'status':'succeeded','receipt':receipt})
return self.inspect(claim.identity)
def mark_review(self,identity):
identity=_identity(identity)
with self._lock:
record=self._read(identity)
if not record or record['status']!='started':raise RecordError('NOT_STARTED')
self._save(identity,{**record,'status':'needs_review'})
return self.inspect(identity)
def inspect(self,identity):
identity=_identity(identity)
with self._lock:
record=self._read(identity)
return {'identity':list(identity),'status':record['status'] if record else 'absent','receipt':record['receipt'] if record else None,'executed_here':False}
An input digest is not an anonymized original
Inputs are compared using SHA-256 of sorted JSON, without storing the original text. Key order does not matter, while exact whitespace does. The caller supplies the definition digest, so the module cannot authenticate that it came from the latest actual code. Fingerprint comparison still depends on trustworthy inputs.
Not storing plaintext does not hide every fact: low-entropy candidates can be guessed and compared. Labels and receipts also need appropriate content handling. Public examples use synthetic values only, with no customer material or credentials. Absence of raw text and complete confidentiality are different claims and should not be presented as equivalent.
Twenty-four tests inspect what remains after a write failure
Tests reopen completed records through another instance and refuse unresolved ones. Changed input and definitions, mismatched identity, invalid JSON, unknown fields and stale claims have checks. Injected replacement failures require no claim after initial save failure and retention of started state after final save failure.
Eight threads using one instance must receive only one initial claim. The package procedure runs twenty-four tests again after extraction. The selector reads five records generated through actual local file examples. These checks establish neither Jupyter-kernel recovery nor external execution or multiprocess safety, and they do not rerun the original browser tests.
Real integration still needs outcome verification and a storage owner
Code, tests, synthetic examples, record generation and README permit modification and reuse. Example and test scratch directories are created beside the package and cleaned afterward. Reopening files is actually checked; operational disaster recovery and every storage device's durability are not established.
Production integration needs an owned writer or suitable database, current authorization and service-result reconciliation. There is no automatic conversion from review-needed to replay. Giving uncertain work a new identity does not make it safe. The module preserves inspectable facts before the next operation is considered, rather than promising universal recovery.