대화와 커널이 바뀌어도 어느 항목을 처리했는지 남기기 위해 독립 파일 원장 예제를 만들었다. 실제 업무를 실행하지 않고 실행 의도·선언한 결과·확인 필요 상태를 저장한다. 파일을 다시 읽는 시험과 저장 실패 시험을 통해 기록이 무엇을 보존하고 무엇을 보장하지 않는지 함께 공유한다.
네 가지 이름으로 기록의 소속을 정한다
프로젝트 범위와 프로그램, 업무 묶음과 항목을 네 개의 짧은 식별자로 받는다. 파일 이름은 이 조합의 hash로 만들고 원장 안에서도 같은 identity인지 대조한다. 사용자가 준 문자열을 그대로 하위 경로로 이어 붙이지 않는다. 저장 위치는 호출자가 지정한 절대 경로의 신뢰할 수 있는 폴더다.
식별자는 실제 계정 소유를 인증하는 증거가 아니다. 같은 폴더를 여러 프로세스가 동시에 쓰는 기능도 없다. 한 인스턴스의 잠금이 짧은 변경 순서만 보호한다. 이 예제를 서버의 다중 사용자 업무 원장이나 운영체제 샌드박스로 소개하지 않고 로컬 기록의 원리를 검토하는 부품으로 공유한다.
실행 의도를 먼저 저장한 뒤 증표를 돌려준다
begin은 입력과 정의 지문을 검사하고 started 상태를 파일에 저장한 뒤 Claim을 돌려준다. 실제 작업을 실행하는 콜백은 받지 않는다. 연결기는 이 반환 전에 외부 행동을 해서는 안 된다는 순서 계약을 따로 적용해야 한다. 원장 함수가 그 바깥의 코드를 강제로 통제하는 것은 아니다.
저장은 고유한 임시 파일에 기록하고 flush와 fsync 뒤 replace하는 방식이다. 실패하면 성공 증표를 반환하지 않고 남은 임시 파일을 정리한다. 저장장치의 모든 전원 실패와 디렉터리 동기화까지 보장하는 데이터베이스 트랜잭션은 아니다. 실제로 구현한 파일 교체의 범위를 명확히 남겼다.
완료와 중단을 다시 읽는 규칙이 다르다
같은 identity에 입력과 정의가 맞고 succeeded 기록이 있으면 보관한 receipt를 reuse로 돌려준다. started나 needs_review면 REVIEW_REQUIRED로 거절한다. 커널이 다시 시작됐다는 이유로 예전 미확정 실행을 새 실행으로 바꾸지 않는다. mark_review는 불확실함을 드러내는 전환이며 완료로 올리는 기능이 아니다.
complete는 현재 Claim의 ID와 입력·정의 지문을 대조하고 호출자가 준 짧은 receipt를 저장한다. 이 receipt도 실제 원격 성공을 확인한 값이라는 보장은 없다. 모듈의 executed_here는 항상 거짓이다. 외부 저장과 게시를 직접 확인하는 층이 없으므로 선언한 결과와 검증한 업무 결과를 같은 말로 표시하지 않는다.
코드와 구현 살펴보기
독립 합성 전체 구현
"""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}
입력 지문도 익명화된 원본은 아니다
입력은 정렬한 JSON의 SHA-256으로 비교하고 원문은 파일에 쓰지 않는다. 키 순서가 달라도 같은 값이면 같고 문자열 공백이 달라지면 다른 입력이다. 정의 hash는 호출자가 전달하므로 그 값이 실제 최신 코드에서 나온 것인지까지 인증하지 않는다. 지문 비교에도 신뢰할 입력의 출처가 필요하다.
hash가 원문을 저장하지 않는다고 모든 정보가 숨겨지는 것은 아니다. 후보가 적은 입력은 추측한 값과 대조할 수 있다. 식별자와 receipt에도 민감한 자료를 넣지 않아야 한다. 공개 예제는 모두 합성 값이며 실제 업체 자료와 인증 값은 사용하지 않았다. 원문 미저장과 완전한 비밀 보장은 다른 주장이다.
24개 시험은 저장 실패 뒤 남는 자료도 본다
새 인스턴스로 완료 기록을 다시 읽어 재사용하고 미완료는 거절하는지 시험했다. 입력과 정의 변경, 잘못된 identity와 JSON, 예상 밖 필드와 오래된 Claim도 확인한다. 파일 교체 오류를 주입해 최초 저장 실패는 증표를 주지 않고 최종 저장 실패는 기존 started를 보존하는지 본다.
여덟 스레드의 같은 인스턴스 시작 요청은 하나만 Claim을 받아야 한다. 전체24개 시험과 ZIP 추출본24개 시험을 수행한다. 아래 선택기는 실제 로컬 파일 예제로 생성한 다섯 기록을 읽는다. 새 Jupyter 커널과 외부 사이트 실행, 여러 프로세스 경합을 검수한 결과는 아니다. 원래 북극성 시험도 다시 실행하지 않았다.
실제 업무 연결에는 결과 확인과 저장 책임이 남는다
코드와 시험, 합성 예제와 기록 생성기, README를 수정·재사용할 수 있다. 예제와 시험의 임시 폴더는 패키지 옆에 만들고 종료 후 정리한다. 파일을 다시 여는 동작은 실제로 확인했지만 운영 중 장애 복구와 저장 매체 전체의 내구성까지 완료했다고 말하지 않는다.
실제 적용에는 한 명의 저장 주체나 적절한 데이터베이스, 현재 권한과 서비스 결과의 재확인이 필요하다. 이 예제에는 확인 필요를 자동 재실행으로 바꾸는 기능이 없다. 불확실한 일을 새 identity로 돌린다고 안전해지는 것도 아니다. 나는 남은 일을 빨리 시작하기 전에 원래 남은 사실을 읽을 수 있게 하는 작은 기록 구조로 공유한다.