#!/usr/bin/env python3 """Local exact-term work planning. No network, credentials or task execution. Uses the existing wire validator and resource ledger. Hypothetical attempts stay fully exposed, including lost replies: estimates are not verified actual usage. A valid plan/request is not authority, publication, reservation or server acceptance. Run unattended with template / assess / request; outputs are exclusive mode 0600. """ from __future__ import annotations import argparse import copy import datetime as dt import hashlib import json import os from pathlib import Path import re import stat import sys import tempfile import concresca_work_budget as budget import concresca_work_client as wire SCHEMA = 'concresca.work_plan.v1' REPORT_SCHEMA = 'concresca.work_plan_assessment.v1' MAX_BYTES = 32768 PHASES = ('discovery', 'negotiation', 'execution', 'verification', 'recovery', 'cleanup') UTC = dt.timezone.utc DAY_MS = 86400000 CAPABILITY = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.:+ -]{0,95}') ID = re.compile(r'[a-z][a-z0-9_-]{0,31}') STEPS = {'stepId','phase','attempts','requestsPerAttempt','maxMillisecondsPerAttempt'} TERM_KEYS = {'schemaVersion','title','purpose','requiredCapabilities','inputs','deliverable', 'checks','actionScope','resourceBudget','value','audience','expiresAt','retentionUntil','publicationConsent'} class PlanError(ValueError): def __init__(self, code: str, field: str = 'plan'): self.code, self.field = code, field super().__init__(code) def need(condition, code, field='plan'): if not condition: raise PlanError(code, field) def fields(value, keys, field): need(type(value) is dict and set(value) == keys, 'fields_invalid', field) def integer(value, lo, hi, field): need(type(value) is int and lo <= value <= hi, 'integer_out_of_range', field) def canonical(value): try: return wire._canonical(value, MAX_BYTES) except wire.WorkClientError: raise PlanError('json_shape_invalid') from None def digest(value): return hashlib.sha256(canonical(value)).hexdigest() def timestamp(value): need(type(value) is str and re.fullmatch(r'[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z', value) is not None, 'utc_timestamp_invalid', 'terms') try: parsed = dt.datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=UTC) need(parsed.year >= 1970, 'utc_timestamp_invalid', 'terms') return int(parsed.timestamp() * 1000) except (ValueError, OverflowError): raise PlanError('utc_timestamp_invalid', 'terms') from None def stamp(ms): return dt.datetime.fromtimestamp(ms / 1000, UTC).isoformat(timespec='seconds').replace('+00:00', 'Z') def reference(value): # Deliberately narrower authoring subset than runtime: ASCII DNS host, HTTPS, # no userinfo/fragment/backslash. Nothing dereferences a selected input. need(type(value) is str and 0 < len(value) <= 1024 and re.fullmatch(r'https://([A-Za-z0-9.-]+)(?::([0-9]{1,5}))?(?:[/?][\x21-\x7e]*)?', value) is not None and '#' not in value and '\\' not in value, 'reference_subset_invalid', 'terms.inputs') hostport = value[8:].split('/', 1)[0].split('?', 1)[0] host, _, port = hostport.partition(':') need(len(host) <= 253 and all(re.fullmatch(r'[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?', x) for x in host.split('.')) and (not port or 1 <= int(port) <= 65535), 'reference_subset_invalid', 'terms.inputs') def validate_plan(value): raw = canonical(value) need(len(raw) <= MAX_BYTES, 'plan_too_large') fields(value, {'schemaVersion','terms','cleanupReserve','operationSlots','steps'}, 'plan') need(value['schemaVersion'] == SCHEMA, 'plan_schema_invalid') terms = value['terms'] fields(terms, TERM_KEYS, 'terms') need(type(terms['publicationConsent']) is bool, 'consent_invalid', 'terms.publicationConsent') # Inspect draft structure without granting consent or changing exported bytes. structural = copy.deepcopy(terms) structural['publicationConsent'] = True try: wire.validate_request('POST', wire.BASE, {'schemaVersion': wire.OPPORTUNITY_SCHEMA, 'terms': structural}) except wire.WorkClientError: raise PlanError('terms_invalid', 'terms') from None need(len(canonical(terms)) <= 12000, 'terms_too_large', 'terms') for label in terms['requiredCapabilities']: need(CAPABILITY.fullmatch(label) is not None, 'capability_subset_invalid', 'terms.requiredCapabilities') for item in terms['inputs']: reference(item['uri']) timestamp(terms['expiresAt']); timestamp(terms['retentionUntil']) fields(value['cleanupReserve'], {'requests','runtimeMilliseconds'}, 'cleanupReserve') integer(value['cleanupReserve']['requests'], 0, 10000, 'cleanupReserve.requests') integer(value['cleanupReserve']['runtimeMilliseconds'], 1, DAY_MS, 'cleanupReserve.runtimeMilliseconds') fields(value['operationSlots'], {'total','cleanup'}, 'operationSlots') integer(value['operationSlots']['total'], 2, 64, 'operationSlots.total') integer(value['operationSlots']['cleanup'], 1, value['operationSlots']['total'] - 1, 'operationSlots.cleanup') steps = value['steps'] need(type(steps) is list and 2 <= len(steps) <= 16, 'steps_invalid', 'steps') ids = set(); count = 0 for i, step in enumerate(steps): field = 'steps.' + str(i) fields(step, STEPS, field) need(type(step['stepId']) is str and ID.fullmatch(step['stepId']) is not None and step['stepId'] not in ids, 'step_id_invalid', field) ids.add(step['stepId']) need(type(step['phase']) is str and step['phase'] in PHASES, 'phase_invalid', field) integer(step['attempts'], 1, 64, field + '.attempts') integer(step['requestsPerAttempt'], 0, 10000, field + '.requestsPerAttempt') integer(step['maxMillisecondsPerAttempt'], 1, DAY_MS, field + '.maxMillisecondsPerAttempt') count += step['attempts'] need(count <= 64, 'attempt_limit_exceeded', 'steps') need(any(s['phase'] == 'cleanup' for s in steps) and any(s['phase'] != 'cleanup' for s in steps), 'cleanup_and_work_required', 'steps') return copy.deepcopy(value) def assess_plan(value, *, now_ms=None): """Worst-case local admission rehearsal, NOT an actual running task ledger.""" now_ms = int(dt.datetime.now(UTC).timestamp() * 1000) if now_ms is None else now_ms integer(now_ms, 0, 253402300799000, 'now') plan = validate_plan(value); terms = plan['terms'] expiry, retention = timestamp(terms['expiresAt']), timestamp(terms['retentionUntil']) reasons = [] def reason(code): if code not in reasons: reasons.append(code) if not (now_ms < expiry <= now_ms + 30 * DAY_MS): reason('work_expiry_invalid') if not (expiry <= retention <= now_ms + 365 * DAY_MS): reason('work_retention_invalid') if terms['publicationConsent'] is not True: reason('publication_consent_missing') limits = {'requests': terms['resourceBudget']['maxRequests'], 'runtimeMilliseconds': terms['resourceBudget']['maxRuntimeSeconds'] * 1000} scope = digest({'schemaVersion': SCHEMA, 'kind': 'hypothetical-local-plan', 'plan': plan}) report = {'schemaVersion': REPORT_SCHEMA, 'checkedAt': stamp(now_ms), 'planSha256': digest(plan), 'termsSha256': digest(terms), 'planningOnly': True, 'executionPerformed': False, 'publicationPerformed': False, 'authorityGranted': False, 'participantEvaluation': None, 'limits': limits, 'cleanupReserve': plan['cleanupReserve'], 'scheduledExposure': {'work':{'requests':0,'runtimeMilliseconds':0,'attempts':0}, 'cleanup':{'requests':0,'runtimeMilliseconds':0,'attempts':0}}, 'attempts': [], 'remainingCapacity': None, 'boundary': 'Worst-case selected plan, not measured usage, a durable task ledger, identity-lifetime accounting or server acceptance. Unknown outcomes release no planned capacity. Each physical retry is separately charged. Cleanup cannot be borrowed. No external tariff or authority is inferred.'} for step in plan['steps']: pool = 'cleanup' if step['phase'] == 'cleanup' else 'work' total = report['scheduledExposure'][pool] total['requests'] += step['attempts'] * step['requestsPerAttempt'] total['runtimeMilliseconds'] += step['attempts'] * step['maxMillisecondsPerAttempt'] total['attempts'] += step['attempts'] ledger = None try: ledger = budget.create_ledger(scope_digest=scope, limits=limits, cleanup_reserve=plan['cleanupReserve'], deadline_ms=expiry, operation_limit=plan['operationSlots']['total'], cleanup_operation_slots=plan['operationSlots']['cleanup']) except budget.BudgetError as e: reason(e.code) if ledger is not None: for i, step in enumerate(plan['steps']): pool = 'cleanup' if step['phase'] == 'cleanup' else 'work' units = {'requests':step['requestsPerAttempt'], 'runtimeMilliseconds':step['maxMillisecondsPerAttempt']} for attempt in range(1, step['attempts'] + 1): operation = f'plan-{i:02d}-{attempt:02d}' binding = digest({'scope':scope,'step':step,'attempt':attempt}) row = {'stepId':step['stepId'],'phase':step['phase'],'attempt':attempt,'pool':pool,'reservedMaximum':units.copy()} try: ledger = budget.reserve(ledger, scope_digest=scope, operation_id=operation, binding_digest=binding, units=units, now_ms=now_ms, pool=pool) ledger = budget.mark_unknown(ledger, scope_digest=scope, operation_id=operation, binding_digest=binding) row.update(admissible=True, reasonCode=None) except budget.BudgetError as e: reason(e.code); row.update(admissible=False, reasonCode=e.code) report['attempts'].append(row) total = budget.totals(ledger, scope_digest=scope) report['remainingCapacity'] = {p:{**total[p]['available'],'operationSlots':total['operationSlotsAvailable'][p]} for p in budget.POOLS} report['readyToPrepareRequest'] = not reasons report['reasonCodes'] = reasons report['status'] = 'plan_ready' if not reasons else 'plan_blocked' return report def compile_request(value, *, now_ms=None): report = assess_plan(value, now_ms=now_ms) need(report['readyToPrepareRequest'], 'plan_not_ready') plan = validate_plan(value) body = {'schemaVersion':wire.OPPORTUNITY_SCHEMA, 'terms':plan['terms']} descriptor = wire.validate_request('POST', wire.BASE, body) return {'schemaVersion':'concresca.prepared_work_request.v1', 'preparedAt':report['checkedAt'], 'planSha256':report['planSha256'],'termsSha256':report['termsSha256'], 'method':descriptor['method'],'path':descriptor['path'],'body':descriptor['body'], 'sent':False,'authorityGranted':False, 'dispatchRequirements':['Current authenticated requester and project scope', 'Recheck current contract, expiry, authority and publication rights at dispatch', 'Use existing WorkController protected custody and cumulative accounting', 'Persist exact body and original idempotency key before dispatch; never reset unknown exposure']} def template(kind='contract-comparison', *, now_ms=None): need(kind in ('contract-comparison','artifact-check'), 'template_invalid') now_ms = int(dt.datetime.now(UTC).timestamp() * 1000) if now_ms is None else now_ms plan = {'schemaVersion':SCHEMA,'terms':{'schemaVersion':'concresca.work_terms.v1', 'title':'Compare public work contract fields' if kind=='contract-comparison' else 'Verify a selected artifact digest', 'purpose':'Compare selected documented fields with current public capabilities; report differences without treating them as workflow acceptance.' if kind=='contract-comparison' else 'Verify the bytes of an explicitly selected local artifact against an expected digest. Do not upload or execute it.', 'requiredCapabilities':['schema-check'] if kind=='contract-comparison' else ['sha256-check'], 'inputs':[], 'deliverable':{'mediaType':'application/json','maxBytes':65536}, 'checks':[{'checkId':'source-digests','description':'Include the selected source byte digests and distinguish observation from acceptance.'}], 'actionScope':['Read only the selected inputs. Do not execute contributed code or publish results automatically.'], 'resourceBudget':{'maxRequests':16,'maxRuntimeSeconds':60}, 'value':{'kind':'uncompensated','description':'No payment or settlement is authorized.'}, 'audience':{'kind':'public'},'expiresAt':stamp(now_ms+DAY_MS),'retentionUntil':stamp(now_ms+2*DAY_MS), 'publicationConsent':False}, 'cleanupReserve':{'requests':3,'runtimeMilliseconds':6000},'operationSlots':{'total':16,'cleanup':3}, 'steps':[{'stepId':'discover','phase':'discovery','attempts':2,'requestsPerAttempt':1,'maxMillisecondsPerAttempt':5000}, {'stepId':'perform','phase':'execution','attempts':1,'requestsPerAttempt':0,'maxMillisecondsPerAttempt':15000}, {'stepId':'verify','phase':'verification','attempts':2,'requestsPerAttempt':1,'maxMillisecondsPerAttempt':5000}, {'stepId':'recover','phase':'recovery','attempts':1,'requestsPerAttempt':1,'maxMillisecondsPerAttempt':5000}, {'stepId':'cleanup','phase':'cleanup','attempts':1,'requestsPerAttempt':1,'maxMillisecondsPerAttempt':3000}]} if kind == 'artifact-check': plan['terms']['checks'] = [{'checkId':'byte-identity','description':'Record expected and measured SHA-256 and byte count; a matching hash alone does not establish correctness.'}] for step in plan['steps']: step['requestsPerAttempt']=0 return plan def _pairs(pairs): result = {} for key, value in pairs: need(key not in result, 'duplicate_json_key') result[key] = value return result def loads_plan(raw): need(type(raw) is bytes and len(raw) <= MAX_BYTES, 'plan_too_large') try: value = json.loads(raw.decode('utf-8'), object_pairs_hook=_pairs, parse_constant=lambda _: (_ for _ in ()).throw(PlanError('json_shape_invalid'))) except (UnicodeError, ValueError, RecursionError) as e: if isinstance(e, PlanError): raise raise PlanError('json_invalid') from None return validate_plan(value) def read_selected(path): fd = os.open(path, os.O_RDONLY | getattr(os, 'O_NOFOLLOW', 0) | getattr(os, 'O_NONBLOCK', 0)) with os.fdopen(fd, 'rb') as f: need(stat.S_ISREG(os.fstat(f.fileno()).st_mode), 'input_not_regular') raw = f.read(MAX_BYTES + 1) return loads_plan(raw) def write_new(path, value): """Complete mode-0600 file, atomically linked into a never-overwritten name. A failed write never publishes a partial target. Parent directory selection remains the local caller's responsibility; no hosting path is inferred. """ raw = (json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + '\n').encode('utf-8') target = Path(path) fd, temporary = tempfile.mkstemp(prefix='.concresca-plan-', dir=str(target.parent)) try: with os.fdopen(fd, 'wb') as f: f.write(raw); f.flush(); os.fsync(f.fileno()) os.link(temporary, target, follow_symlinks=False) finally: os.unlink(temporary) def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest='command', required=True) t = sub.add_parser('template'); t.add_argument('--kind', choices=['contract-comparison','artifact-check'],default='contract-comparison');t.add_argument('--output',required=True) for command in ('assess','request'): p = sub.add_parser(command);p.add_argument('plan');p.add_argument('--output',required=True) args = parser.parse_args(argv) try: if args.command == 'template': result = template(args.kind) else: plan = read_selected(args.plan) result = assess_plan(plan) if args.command == 'assess' else compile_request(plan) write_new(args.output, result) print(json.dumps({'status':'written','command':args.command,'networkRequests':0,'executionPerformed':False, 'readyToPrepareRequest':result.get('readyToPrepareRequest')})) return 2 if result.get('readyToPrepareRequest') is False else 0 except PlanError as e: print(json.dumps({'status':'blocked','reasonCode':e.code,'field':e.field,'executionPerformed':False}),file=sys.stderr) return 2 except (OSError, ValueError, OverflowError): print(json.dumps({'status':'blocked','reasonCode':'local_io_error','executionPerformed':False}),file=sys.stderr) return 3 if __name__ == '__main__': sys.exit(main())