#!/usr/bin/env python3 """Code4History two-way workbench. All operations are local, stdlib-only. Use --root to target a test or separate project. No daemon or background process. This tool safely archives upload bytes and moves completed human/agent tasks; semantic historical interpretation remains the responsibility of the research skill. """ from __future__ import annotations import argparse from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import re import shutil import sys import tempfile ROOT = Path(__file__).resolve().parents[1] SRC_KINDS = {'primary': ('SRC-P', 'primary'), 'secondary': ('SRC-S', 'secondary')} WORK_FILES = {'readme.md', '.gitkeep'} SIDE_SUFFIX = '.source.md' TEMP_SUFFIXES = ('.part', '.tmp', '.crdownload', '.download', '.swp') def stamp(): return datetime.now(timezone.utc).isoformat(timespec='seconds') def sha256(path): d = hashlib.sha256() with path.open('rb') as f: for part in iter(lambda: f.read(1024*1024), b''): d.update(part) return d.hexdigest() def safe_file(root: Path, relative: str, *, required=True) -> Path: # Never accept absolute, dotted segments, traversal, directory / symlink. if not relative or '\\' in relative or '/' in relative or relative in {'.', '..'} or Path(relative).is_absolute(): raise ValueError('Only a direct filename inside the target inbox is allowed.') p = root / relative if required and (not p.is_file() or p.is_symlink()): raise ValueError(f'Not a regular file: {p}') if p.exists() and p.is_symlink(): raise ValueError(f'Unsafe symbolic link: {p}') return p def atomic_bytes(dest: Path, content: bytes): dest.parent.mkdir(parents=True, exist_ok=True) if dest.exists(): raise FileExistsError(dest) with tempfile.NamedTemporaryFile(dir=dest.parent, prefix='.c4h-incoming-', delete=False) as f: temp = Path(f.name) f.write(content) f.flush() os.fsync(f.fileno()) try: os.replace(temp, dest) finally: temp.unlink(missing_ok=True) def atomic_replace(dest: Path, content: bytes): dest.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile(dir=dest.parent, prefix='.c4h-write-', delete=False) as f: temp=Path(f.name); f.write(content); f.flush(); os.fsync(f.fileno()) try: os.replace(temp, dest) finally: temp.unlink(missing_ok=True) def safe_copy(source: Path, destination: Path, digest: str): destination.parent.mkdir(parents=True,exist_ok=True) if destination.exists(): if sha256(destination) == digest: return False raise FileExistsError(f'Destination exists with different bytes: {destination}') with tempfile.NamedTemporaryFile(dir=destination.parent,prefix='.c4h-upload-',delete=False) as f: temp=Path(f.name) try: shutil.copyfile(source,temp) if sha256(temp) != digest: raise IOError(f'Hash mismatch while staging: {source}') os.replace(temp,destination) finally: temp.unlink(missing_ok=True) if sha256(destination) != digest: raise IOError(f'Hash mismatch after archive write: {destination}') return True def user_answer(doc: str) -> str: if '## 用户回复' not in doc: return '' reply = doc.split('## 用户回复',1)[1].split('## Agent 处理记录',1)[0].strip() if not reply or reply.startswith('（请在此填写'): return '' return reply def status(root: Path) -> dict: upl=root/'upload' files=[p for p in sorted(upl.iterdir()) if p.name not in WORK_FILES] if upl.exists() else [] attachments=[]; source_notes=[]; blocked=[] for p in files: if p.is_symlink() or not p.is_file(): blocked.append(p.name);continue if p.name.startswith('.') or p.name.lower().endswith(TEMP_SUFFIXES): blocked.append(p.name);continue if p.name.endswith(SIDE_SUFFIX): source_notes.append(p.name);continue companion=p.name+SIDE_SUFFIX attachments.append({'file':p.name,'source_note':companion if (upl/companion).is_file() else None,'bytes':p.stat().st_size}) paired={x['source_note'] for x in attachments if x['source_note']} todo=[] for p in sorted((root/'todo').glob('TODO-*.md')): if p.is_symlink(): blocked.append('todo/'+p.name);continue body=p.read_text(encoding='utf-8') m=re.search(r'^status:\s*([a-z_-]+)\s*$',body,re.M) state=m.group(1) if m else 'invalid' if state=='open' and user_answer(body):state='answered' todo.append({'file':p.name,'status':state}) req=[] for p in sorted((root/'requests').iterdir()) if (root/'requests').exists() else []: if p.name in WORK_FILES or p.name=='archive':continue if p.is_file() and not p.is_symlink():req.append(p.name) else:blocked.append('requests/'+p.name) return { 'upload':attachments, 'orphan_source_notes':sorted(set(source_notes)-paired), 'blocked':blocked, 'todo':todo, 'requests':req, 'summary':{ 'upload_pending':len(attachments)+len(set(source_notes)-paired)+len([x for x in blocked if '/' not in x]), 'todo_open':sum(x['status']=='open' for x in todo), 'todo_answered':sum(x['status']=='answered' for x in todo), 'requests_pending':len(req) } } def next_number(folder: Path, prefix: str) -> int: seq=[] for p in folder.glob(f'{prefix}-*.md'): m=re.match(re.escape(prefix)+r'-(\d{4})(?:-|\.md)',p.name) if m:seq.append(int(m.group(1))) return max(seq,default=0)+1 def next_source_id(root: Path,kind: str): prefix, sub=SRC_KINDS[kind] folder=root/'01_sources/records'/sub return f'{prefix}-{next_number(folder,prefix):04d}' def yaml_set(doc: str,key: str,val: str) -> str: value=json.dumps(val,ensure_ascii=False) pattern=re.compile(r'^'+re.escape(key)+r':.*$',re.M) if pattern.search(doc.split('\n---\n',1)[0]): return pattern.sub(lambda _:f'{key}: {value}',doc,count=1) return doc.replace('\n---\n',f'\n{key}: {value}\n---\n',1) def intake(root:Path,filename:str,kind:str,slug:str,dry_run=False): if not re.fullmatch(r'[a-z0-9]+(?:-[a-z0-9]+)*',slug): raise ValueError('slug must be lowercase ASCII kebab-case') if filename in WORK_FILES or filename.endswith(SIDE_SUFFIX) or filename.startswith('.') or filename.lower().endswith(TEMP_SUFFIXES): raise ValueError('Not an upload attachment filename') inbox=root/'upload' source=safe_file(inbox,filename) if source.stat().st_size==0: raise ValueError('Zero-byte upload must be checked, not archived') metadata=safe_file(inbox,filename+SIDE_SUFFIX,required=False) has_meta=metadata.is_file() digest=sha256(source); note_hash=sha256(metadata) if has_meta else None # A previous round may have been interrupted after successful archival but # before upload cleanup, or the user may have uploaded identical bytes. # Reuse an existing verified accession; NEVER discard a changed sidecar. for prior in sorted((root/'01_sources/provenance').glob('SRC-*-intake.json')): try: receipt_old=json.loads(prior.read_text(encoding='utf-8')) archived=root/receipt_old['archive_path'] if (receipt_old.get('sha256')!=digest or not archived.is_file() or sha256(archived)!=digest): continue if receipt_old.get('source_note_sha256')!=note_hash: raise ValueError('已归档相同原件，但此次来源说明不同；保留上传件供人工合并新的来源信息。') record_id=receipt_old['source_id'] record_matches=list((root/'01_sources/records'/SRC_KINDS[kind][1]).glob(record_id+'-*.md')) if not record_id.startswith(SRC_KINDS[kind][0]+'-') or not record_matches: continue if dry_run:return {'id':record_id,'reused':True,'sha256':digest} if sha256(source)!=digest or (has_meta and sha256(metadata)!=note_hash): raise IOError('Upload changed while checking duplicate') source.unlink() if has_meta:metadata.unlink() return {'id':record_id,'reused':True,'sha256':digest} except (KeyError, json.JSONDecodeError): continue ident=next_source_id(root,kind) suffix=Path(filename).suffix.lower() original=root/'01_sources/originals'/SRC_KINDS[kind][1]/f'{ident}--{filename}' meta_dest=root/'01_sources/provenance'/f'{ident}-upload-source.md' receipt=root/'01_sources/provenance'/f'{ident}-intake.json' record=root/'01_sources/records'/SRC_KINDS[kind][1]/f'{ident}-{slug}.md' if any(p.exists() for p in (original,receipt,record,meta_dest)): raise FileExistsError('Generated destination is occupied; inspect before retrying') plan={'id':ident,'original_path':original.relative_to(root).as_posix(),'record_path':record.relative_to(root).as_posix(),'metadata_path':meta_dest.relative_to(root).as_posix() if has_meta else None,'sha256':digest,'metadata_sha256':note_hash,'metadata_missing':not has_meta} if dry_run:return plan # Validate all inputs, prepare copies and documentation before removing originals. template=root/'10_templates'/('primary.md' if kind=='primary' else 'secondary.md') card=template.read_text(encoding='utf-8').replace('__ID__',ident) for field,value in [('original_path',plan['original_path']),('sha256',digest),('original_filename',filename),('ingested_utc',stamp()),('upload_source_note',plan['metadata_path'] or ''),('source_metadata_status','provided_unverified' if has_meta else 'missing')]: card=yaml_set(card,field,value) # Staging and creating records may fail. Until receipt + record + archive checks pass, # never remove the user's source. Partial copies are left for recovery. safe_copy(source,original,digest) if has_meta:safe_copy(metadata,meta_dest,note_hash) atomic_bytes(record,card.encode('utf-8')) receipt_data={'source_id':ident,'timestamp_utc':stamp(),'upload_name':filename,'archive_path':plan['original_path'],'sha256':digest,'source_note_path':plan['metadata_path'],'source_note_sha256':note_hash,'ingestion':'archived_not_necessarily_read'} atomic_bytes(receipt,(json.dumps(receipt_data,ensure_ascii=False,indent=2)+'\n').encode('utf-8')) if sha256(original)!=digest or not record.is_file() or not receipt.is_file() or (has_meta and sha256(meta_dest)!=note_hash): raise IOError('Archive integrity check failed; upload retained') # Raw catalog updates can be retried by Agent if interrupted; receipt is durable. catalog=root/'01_sources/catalog.md' current=catalog.read_text(encoding='utf-8') if catalog.exists() else '# Source catalog\n' line=f'\n- `{ident}` · `{plan["original_path"]}` · sha256 `{digest}` · [SRC card]({record.relative_to(root).as_posix()}) · metadata: {"provided" if has_meta else "pending"}\n' atomic_replace(catalog,(current+line).encode('utf-8')) # Recheck bytes before removing inbox: if researcher modified during intake, retain. if sha256(source)!=digest or (has_meta and sha256(metadata)!=note_hash): raise IOError('Upload changed during intake; retaining input to avoid data loss') # At this stage all copies & receipt are verified; unlink originals only now. source.unlink() if has_meta:metadata.unlink() return plan def todo_new(root:Path,title:str,kind:str,body:str,ref:str,dry_run=False): if not title.strip() or not body.strip():raise ValueError('TODO title and body required') n=next_number(root/'todo','TODO');ident=f'TODO-{n:04d}' doc=f'''--- id: {ident} status: open kind: {kind} created: "{stamp()}" reference: {json.dumps(ref,ensure_ascii=False)} --- # {title} ## 为什么需要用户 {body} ## 用户回复 （请在此填写回复；扫描器会自动识别） ## Agent 处理记录 （处理后说明关联的正式研究记录） ''' dest=root/'todo'/f'{ident}-request.md' if not dry_run:atomic_bytes(dest,doc.encode('utf-8')) return str(dest.relative_to(root)) def archive_task(root:Path,folder:str,name:str,record:str,require_answer=False,dry_run=False): source=safe_file(root/folder,name) if source.name=='readme.md' or not source.name.endswith('.md'): raise ValueError('Expected a Markdown task file') evidence=(root/record).resolve() if not evidence.is_file() or (root.resolve()!=evidence and root.resolve() not in evidence.parents): raise ValueError('The --record path must be an existing file in this research repository') if folder=='todo': if not re.fullmatch(r'TODO-\d{4}-[a-zA-Z0-9_-]+\.md',name): raise ValueError('Invalid TODO task file') doc=source.read_text('utf-8') status_line=re.search(r'^status:\s*([a-z_-]+)\s*$',doc,re.M) if not status_line or status_line.group(1) not in ('open','answered') or not user_answer(doc): raise ValueError('TODO needs an actual user answer in ## 用户回复') dest=root/folder/'archive'/name if dest.exists():raise FileExistsError(f'Already archived: {dest}') if dry_run:return str(dest.relative_to(root)) doc=source.read_text('utf-8') if folder=='todo': doc=re.sub(r'^status:\s*open\s*$', 'status: answered', doc, count=1, flags=re.M) applied=f'\n\n## 归档说明\n- 时间：{stamp()}\n- Agent 已处理并记录于：`{record}`\n' # Preserve the original reply while adding an archival statement. atomic_bytes(dest,(doc+applied).encode('utf-8')) if not dest.read_bytes().startswith(doc.encode('utf-8')): raise IOError('Archive content mismatch') source.unlink() return str(dest.relative_to(root)) def finish(root:Path)->int: s=status(root) print(json.dumps(s['summary'],ensure_ascii=False,indent=2)) if s['summary']['upload_pending'] or s['summary']['todo_answered'] or s['summary']['requests_pending']: print('有未完成的交接事项：请更新 handoff/latest.md；未成功归档的上传原件必须保留。') return 1 print('已无待摄入 upload 文件、未处理用户答复或 requests；开放 TODO 可保留等待用户。') return 0 def main(argv=None): p=argparse.ArgumentParser(description=__doc__) p.add_argument('--root',type=Path,default=ROOT) sub=p.add_subparsers(dest='command',required=True) sub.add_parser('scan') intake_p=sub.add_parser('intake');intake_p.add_argument('--file',required=True);intake_p.add_argument('--kind',choices=tuple(SRC_KINDS),required=True);intake_p.add_argument('--slug',required=True);intake_p.add_argument('--dry-run',action='store_true') todo_p=sub.add_parser('todo-new');todo_p.add_argument('--title',required=True);todo_p.add_argument('--kind',choices=('question','verification','decision','authorization','assistance'),default='question');todo_p.add_argument('--body',required=True);todo_p.add_argument('--ref',default='');todo_p.add_argument('--dry-run',action='store_true') close_p=sub.add_parser('todo-close');close_p.add_argument('--id',required=True);close_p.add_argument('--record',required=True);close_p.add_argument('--dry-run',action='store_true') req_p=sub.add_parser('request-done');req_p.add_argument('--file',required=True);req_p.add_argument('--record',required=True);req_p.add_argument('--dry-run',action='store_true') sub.add_parser('finish') args=p.parse_args(argv);root=args.root.resolve() try: if args.command=='scan':print(json.dumps(status(root),ensure_ascii=False,indent=2)) elif args.command=='intake':print(json.dumps(intake(root,args.file,args.kind,args.slug,args.dry_run),ensure_ascii=False,indent=2)) elif args.command=='todo-new':print(todo_new(root,args.title,args.kind,args.body,args.ref,args.dry_run)) elif args.command=='todo-close':print(archive_task(root,'todo',safe_file(root/'todo',next((x.name for x in (root/'todo').glob(args.id+'-*.md')),args.id)).name,args.record,True,args.dry_run)) elif args.command=='request-done':print(archive_task(root,'requests',args.file,args.record,False,args.dry_run)) elif args.command=='finish':return finish(root) except (ValueError,OSError,RuntimeError,StopIteration) as ex: print(f'交接失败：{ex}',file=sys.stderr) return 2 return 0 if __name__=='__main__':sys.exit(main())