"""Selected case journeys derived from module stages, with five reusable UI patterns.""" from pathlib import Path import json from case_content import content_for, narrative HERE=Path(__file__).resolve().parent ARCH=HERE.parent MANIFESTS={k:json.loads((ARCH/'contracts/manifests'/f'{k}.json').read_text()) for k in ('support','data','configuration','source','commercial')} SCREENS=['queue','workspace','decision','result','learning'] CONFIG_STAGES={ 'CF-02':['S2','S3'], 'CF-03':['S3','S5'], 'CF-04':['S2'], 'CF-05':['S2','S4','S5'], 'CF-06':['S2','S3'], 'CF-07':['S3'], 'CF-08':['S5'], 'CF-09':['S5'], 'CF-10':['S3'], 'CF-11':['S5'], 'CF-12':['S2','S3','S4','S5'], 'CF-13':['S4'], 'CF-14':['S2','S3'], 'CF-15':['S3'], 'CF-16':['S2','S3','S4','S5'], 'CF-17':['S2','S3'], 'CF-18':['S3'], 'CF-19':['S2','S4','S5'], 'CF-20':['S5'], 'CF-21':['S3','S5'], 'CF-23':['S3'], 'CF-24':['S3'], 'CF-25':[]} EXTERNAL={ 'DC-02':'Bulstrad IT', 'DC-03':'Customer', 'DC-04':'Bulstrad IT', 'DC-05':'Bulstrad IT', 'DC-12':'Bulstrad IT', 'DC-22':'Authorised customer operator', 'DC-13':'Bulstrad IT', 'DC-16':'Commission owner', 'DC-21':'Authorised customer operator', 'DC-24':'Infrastructure team', 'DC-19':'Authorised customer operator', 'CF-01':'Authorised customer operator', 'CF-06':'INSIS configuration owner', 'CF-08':'Document template owner', 'CF-09':'Numbering owner', 'CF-10':'INSIS configuration owner', 'CF-12':'INSIS configuration owner', 'CF-14':'INSIS configuration owner', 'CF-17':'INSIS product owner', 'CF-21':'Nomenclature owner','DV-02':'Document template owner'} FAILURE_STAGE={ 'DC-01':'S3','DC-07':'S3','DC-10':'S3','DC-19':'S7','DC-20':'S5','DC-22':'S3','CF-01':'S4','CF-22':'S5', 'CF-04':'release','CF-09':'S5','CF-10':'S3','CF-13':'S4','CF-15':'S3','CF-18':'S1','CF-19':'retirement','CF-21':'S3','CF-23':'S3','CF-25':'S1', 'DV-03':'S1','DV-06':'S4','DV-08':'S1','DV-09':'S4','DV-12':'S4','DV-13':'S2','DV-17':'S7','DV-18':'S5','DV-19':'S5'} LINKED={ 'CF-02':('source','Conditional rule implementation'), 'CF-16':('source','Issue process for the product variant'), 'CF-19':('source','Campaign eligibility and duration code'), 'CF-24':('source','Contract-type consumer and product process'), 'DV-02':('configuration','Document registration and template parameters'), 'DV-11':('configuration','Framework contract-type values'), 'DV-15':('configuration','Control activation and message rows')} LABELS={ 'support':{'S1':'Classification','route':'The route','closure':'Closure'}, 'data':{'S3':'Solution take','S2':'Investigation','S4':'Confirmation','S5':'Application','S6':'Validation','S7':'Reversal / restoration','S8':'Precipitation'}, 'configuration':{'S1':'Normalization','S2':'Abacus','S3':'IPAL','S4':'Offer','S5':'Serdica','joint':'Joint proof','release':'Release and verify'}, 'commercial':{'O1':'Assess effort','O2':'Prepare offer','O3':'Customer decision','O4':'Communicate decision'}, 'source':{'S1':'Planning','S2':'Approval','S3':'Implementation','S4':'Validation','S5':'Release scripts','S6':'Deploy approval','S7':'Deploy and verify','S8':'Reversal'}} def definition(module,stage): return next(s for s in MANIFESTS[module]['stages'] if s['id']==stage) def verification_profile(module,stage): if module=='configuration': domain={'S2':'abacus','S3':'ipal','S4':'offer','S5':'serdica'}.get(stage,'integration') return 'configuration.'+domain+'.validator' return 'source.validator' if module=='source' else 'support.validator' def step(module,stage,*,visit='',screen=None,protocol=None,gate=None,actor='approver',label=None,branch='main'): d=definition(module,stage) return {'uid':f'{branch}/{module}/{stage}'+('/'+visit if visit else ''),'module':module,'stage':stage, 'label':LABELS[module][stage],'context_label':label,'screen':screen or 'workspace','protocol':protocol or 'evidence', 'active_gate':gate,'decision_actor':actor,'branch':branch,'profile':d['profile'], 'input_contracts':d['consumes'],'output_contracts':d['produces'],'gate_contracts':d['gates'],'done':d['done']} def main_steps(case,scenario='first'): cid=case['id']; module=case['ui']['working_module']; result=[] if case['mode']=='core': return [step('commercial','O1',protocol='assessment'), step('commercial','O2',screen='decision',protocol='offer_draft',gate='H3'), step('commercial','O3',screen='result',protocol='offer_acceptance',gate='CG-11',actor='customer'), step('commercial','O4',screen='decision',protocol='send',gate='H5-send')] if module=='data': result=[step('data','S3',visit='take')] if scenario!='similar' and case['mode']!='answer' and case['ui']['content'].get('initial_route')!='ready': result += [step('data','S2'),step('data','S3',visit='return',label='Solution take · return',screen='decision',protocol='decision',gate='H2')] result += [step('data','S4',screen='decision',protocol='decision',gate='H3'), step('data','S5',protocol='external' if case['ui']['external_owner'] else 'send' if case['mode']=='answer' else 'execution',screen='workspace' if case['ui']['external_owner'] else 'decision',gate=None if case['ui']['external_owner'] else 'H5-send' if case['mode']=='answer' else 'H5'), step('data','S6',protocol='verification')] if case['ui']['restoration_required']: result.append(step('data','S7',screen='decision',protocol='reversal',gate='H7',label='Restore the temporary exception')) elif module=='configuration': result=[step(module,'S1',screen='decision',protocol='scope',gate='H1',actor='customer')] for sid in case['ui']['selected_stages']: external_sid='S5' if cid in ('CF-08','CF-09','DV-02') else 'S3' waiting=case['ui']['external_owner'] and sid==external_sid result.append(step(module,sid,protocol='external' if waiting else 'stage_result',screen='workspace' if waiting else 'decision',gate=None if waiting else 'H3')) if cid!='CF-25': result += [step(module,'joint',screen='result',protocol='preview',gate='CG-11',actor='customer'), step(module,'release',screen='decision',protocol='release',gate='H5')] elif case['mode']=='programme': result=[step('source','S1',protocol='referral',label='Assess programme scope and ownership')] else: for sid in ['S1','S2','S3','S4','S5','S6','S7']: modes={'S1':('workspace','evidence',None),'S2':('decision','decision','H2'), 'S3':('workspace','implementation',None), 'S4':('decision','review','H3'),'S5':('decision','release_plan','H4'), 'S6':('decision','release','H5'),'S7':('workspace','verification',None)} screen,protocol,gate=modes[sid] result.append(step('source',sid,screen=screen,protocol=protocol,gate=gate)) return result def groups(steps,module): if module!='configuration':return [{'kind':'sequence','steps':[s['uid'] for s in steps]}] head=[s['uid'] for s in steps if s['stage'] in ('S1','S2')] product=[s['uid'] for s in steps if s['stage'] in ('S3','S4')] wiring=[s['uid'] for s in steps if s['stage']=='S5'] tail=[s['uid'] for s in steps if s['stage'] in ('joint','release')] out=[{'kind':'sequence','steps':head}] if product and wiring:out.append({'kind':'parallel','lanes':[product,wiring],'label':'Product/offer and wiring branches can run together'}) elif product or wiring:out.append({'kind':'sequence','steps':product or wiring}) if tail:out.append({'kind':'sequence','steps':tail}) return out def stage_states(s): """Several card states may belong to one stage; these are not new workflow nodes.""" variants=[{**s,'state_label':'Illustrated stage state','navigation_uid':s['uid']}] def add(key,label,screen,protocol,gate=None,actor='approver'): variants.append({**s,'uid':s['uid']+'/state-'+key,'navigation_uid':s['uid'], 'state_label':label,'screen':screen,'protocol':protocol, 'active_gate':gate,'decision_actor':actor}) if s['module']=='configuration' and s['stage'] in ('S2','S3','S4','S5'): add('work','Prepare and iterate','workspace','evidence') add('plan','Plan ready · H2','decision','decision','H2') if s['protocol']!='external': add('apply','Working-target action · effect gate','decision','execution',s.get('working_effect_gate','HW-instance')) else: add('checked','Returned evidence checked · H3','decision','stage_result','H3') if s['protocol']!='external':variants=variants[1:]+variants[:1] elif s['module']=='configuration' and s['stage']=='release' and s['protocol']!='reversal': add('plan','Release plan ready · H4','decision','release_plan','H4') add('checked','Target result checked','workspace','verification') elif s['module']=='support' and s['stage']=='S1': add('classification','Review classification · H1','decision','decision','H1') elif s['module']=='support' and s['stage']=='route': add('handover','Receiving module accepts · Hd','decision','handover','Hd') elif s['module']=='support' and s['stage']=='closure': add('send','Reply ready · send gate','decision','send','H5-send') elif s['module']=='source' and s['stage']=='S1' and s['protocol']!='referral': add('scope','Scope ready · H1','decision','decision','H1') elif s['module']=='data' and s['stage']=='S3' and s['screen']=='workspace': add('plan','Ready solution plan · H2','decision','decision','H2') elif s['module']=='data' and s['stage']=='S5' and s['protocol']=='external': add('send','Instructions ready · send gate','decision','send','H5-send') add('reported','Execution receipt received · verification pending','workspace','external_report') if s.get('has_platform_step'):add('apply','Platform-owned step · effect gate','decision','execution','H5') elif s['module']=='data' and s['stage']=='S5' and s['protocol']=='execution': add('applied','Application outcome · verification follows','workspace','application_result') if s['protocol']=='reversal': add('checked','Restoration or recovery checked','workspace','verification') return variants def derive(cases): specs=json.loads((HERE/'case-ui-specs.json').read_text(encoding='utf-8'))['cases'] result={} for c in cases: cid=c['id']; ui=dict(specs[cid]); ui['content']=content_for(c); module='commercial' if c['mode']=='core' else 'configuration' if cid.startswith('CF') and cid not in ('CF-01','CF-22') else 'source' if cid.startswith('DV') else 'data' ui.update(working_module=module,offer_type='core_development' if c['mode']=='core' else None,selected_stages=CONFIG_STAGES.get(cid,[]),external_owner=EXTERNAL.get(cid), restoration_required=cid in ('DC-19','CF-01'),scheduled_retirement=cid=='CF-19', core_terms_pending=c['mode']=='core',outcome_type='referral' if c['mode'] in ('programme','core') else 'report' if cid=='CF-25' else 'answer' if c['mode']=='answer' else 'operational') ui['routing_exception']=({'title':'CF-01 is a configuration-shaped request, handled as a temporary Support/Data exception.', 'reason':'CF is the catalogue prefix, not the runtime module. This request relaxes an existing rule for a bounded action and restores it. It uses Data S3–S7 under the same Support root. A lasting rule change follows Configuration (CF-02).'} if cid=='CF-01' else {'title':'CF-22 is master-data work under Support/Data.','reason':'BSO batch provisioning uses the Data stages. Changes to blank types or numbering behaviour belong to Configuration; the catalogue includes both request shapes.'} if cid=='CF-22' else None) ui['external_instruction']={'DC-03':'Log in using your own account and report whether the agreed access works.', 'DC-19':'Perform the agreed policy action while the approved exception is active, then report its result.', 'CF-01':'Perform the agreed policy action while the approved exception is active, then report its result.', 'DV-02':'Provide the agreed document template and evidence from the exact printed example.'}.get(cid,ui['outcome']) ui['scope_note'] = ('Assessment & Offers: clarify, assess effort, prepare offer, customer decision and communication. Core is the offer type; owner terms remain pending and no implementation is authorised.' if c['mode']=='core' else 'Programme referral only; implementing the programme is outside the current small-change charter.' if c['mode']=='programme' else 'Report-only comparison; no configuration is applied.' if cid=='CF-25' else 'Temporary exception stays under Support/Data, with required restoration.' if ui['restoration_required'] else 'BSO master-data work stays under Support/Data.' if cid=='CF-22' else 'Selected rating templates; unchanged product catalogue is supplied by a current baseline.' if cid=='CF-05' else 'Tariff-only example; catalogue and wiring changes would extend the approved scope.' if cid=='CF-04' else 'INSIS-native product: rating and interface work are joined with the external product owner.' if cid=='CF-17' else 'The illustrated variant selects only the work required for this requested outcome.') copy={**c,'ui':ui} paths={name:main_steps(copy,name) for name in ('first','similar')} root=[step('support','S1',screen='queue',protocol='intake'),step('support','route',screen='decision',protocol='route',gate='H2')] close=step('support','closure',screen='result',protocol='delivery',gate='CG-11',actor='customer') aux=[] if module=='data' and (c['mode']=='answer' or ui['content'].get('initial_route')=='ready'): aux += [step('data','S2',branch='alternative',label='Investigate when no ready answer applies'), step('data','S3',visit='return',branch='alternative',label='Return the checked answer to solution take',screen='decision',protocol='decision',gate='H2')] if module=='data':aux.append(step('data','S8',screen='learning',protocol='learning',branch='after')) else:aux.append({'uid':'after/learning','module':module,'stage':None,'label':'Post-result learning','screen':'learning','protocol':'learning','active_gate':None,'decision_actor':'publisher','branch':'after','profile':'support.curator','input_contracts':['paper:report'],'output_contracts':['experience-entry'],'gate_contracts':[],'done':'Record actual outcome and propose the coherent knowledge/skill/tool change.'}) if c['mode'] not in ('programme','core','answer') and cid!='CF-25' and not ui['restoration_required']: aux.append(step(module,'S7' if module=='data' else 'S8' if module=='source' else 'release',visit='recovery',screen='decision',protocol='reversal',gate='H7',label='Reconcile and decide reversal',branch='exception')) if ui['scheduled_retirement']: aux.append(step('configuration','release',visit='retirement',screen='decision',protocol='reversal',gate='H7',label='Scheduled campaign retirement',branch='after')) linked=[] if cid in LINKED: lm,scope=LINKED[cid] ids=['S1','S2','S3','S4','S5','S6','S7','S8'] if lm=='source' else {'DV-02':['S1','S5','joint','release'],'DV-11':['S1','S3','joint','release'],'DV-15':['S1','S3','S5','joint','release']}[cid] linked_case={**copy,'mode':'source' if lm=='source' else 'configuration', 'ui':{**ui,'working_module':lm,'selected_stages':[x for x in ids if x in ('S2','S3','S4','S5')], 'external_owner':EXTERNAL.get(cid) if lm=='configuration' else None}} linked=main_steps(linked_case) for s in linked:s['uid']=s['uid'].replace('main/','linked/',1);s['branch']='linked' ui['linked_scope']=scope skipped=[{'stage':sid,'reason':'Outside this example delta; current baseline supplies any required unchanged facts.'} for sid in ['S2','S3','S4','S5'] if module=='configuration' and sid not in ui['selected_stages']] failure_stage=FAILURE_STAGE.get(cid,'S6' if module=='data' else 'joint' if module=='configuration' else 'S8') failure=next((s for s in paths['first']+aux if s['stage']==failure_stage or (failure_stage=='retirement' and s['uid'].endswith('/retirement'))),paths['first'][0]) # A manifest's default profile is a dispatch seam, not a universal specialist. for s in root+[close]+paths['first']+paths['similar']+aux+linked: variants=definition(s['module'],s['stage']).get('profile_variants',{}) if s['stage'] else {} selected=[p for p in c['profiles'] if p in variants.values()] if variants and not selected and s['module']=='data':selected=['support.investigator.master_data'] s['profiles']=selected or [s['profile']] s['profile']=s['profiles'][0] if s['module']=='source' and s['stage']=='S3' and 'source.camunda_developer' in c['profiles']: s['profiles'].append('source.camunda_developer') s['has_platform_step']=cid in ('DC-03','DC-19','CF-01') and s['module']=='data' and s['stage']=='S5' s['working_effect_gate']='HW-shared' if cid in ('CF-03','CF-11','CF-20','CF-21') else 'HW-instance' result[cid]={'ui':ui,'root':root,'closure':close,'paths':paths,'groups':{k:groups(v,module) for k,v in paths.items()},'auxiliary':aux,'linked':linked,'skipped':skipped,'failure_uid':failure['uid'], 'similar_condition':'The published method/tool passes its current applicability check. A mismatch returns to investigation or scope revision; permissions and proof do not disappear.', 'screens':SCREENS,'module_manifest':f'contracts/manifests/{module}.json'} result[cid]['stage_states']={s['uid']:stage_states(s) for s in all_steps(result[cid])} if module in ('data','support','commercial'): result[cid]['stage_states'][root[1]['uid']]=[s for s in result[cid]['stage_states'][root[1]['uid']] if s['protocol']!='handover'] return result def all_steps(journey): values=journey['root']+[journey['closure']]+journey['paths']['first']+journey['paths']['similar']+journey['auxiliary']+journey['linked'] values += [s for group in journey.get('stage_states',{}).values() for s in group] return list({s['uid']:s for s in values}.values()) def bind_protocols(case,journey,fragments,base): """Bind shared transport/evidence operations to actual module stages, never a ten-step flow.""" import copy,hashlib common=fragments mapping={'scope':'delivery','route':'decision','handover':'decision','stage_result':'decision','preview':'delivery','release':'decision', 'implementation':'evidence','test':'verification','review':'decision','release_plan':'decision', 'referral':'evidence','core_request':'evidence','assessment':'evidence','offer_draft':'decision','offer_acceptance':'delivery','external':'execution','external_report':'verification','application_result':'verification','send':'decision','learning':'retention'} out={} for s in all_steps(journey): protocol=s['protocol']; record=copy.deepcopy(common[mapping.get(protocol,protocol)]) record['stage_context']=s record['narrative']=narrative(case,journey,s) record['chat_protocol']={'method':'POST','path':'/cases/{caseId}/commands', 'case_binding':'Use the controlling Support case. Resolve the task inside that case or an accepted, viewer-visible descendant; send to the main root for bounded delegation. Unrelated cases are rejected.', 'headers':{'Idempotency-Key':case['id']+'-'+s['uid'].replace('/','-')+'-chat-steer'}, 'body':{'command':'steer','text':'Regarding '+s['module']+' '+str(s['stage'])+' plan: include the stated negative check before requesting approval.'}, 'response':'202; controller-authenticated steering is persisted and projected as cards. It does not decide a gate.', 'storage':'INBOX_MESSAGES command dedup, case/task context in PAPERS, AUDIT_LEDGER, OUTBOX_MESSAGES; dependent plan review resumes through the root.', 'subplan':'Child Plan -> immediate parent PlanConfirmation -> bounded preparation. Human H2 and effect decisions remain separate; a child does not ask the customer directly.'} record['case_context']={'origin':'Support-origin catalogue example', 'root_case':'fixture-'+case['id'].lower(), 'working_case':'same Support case' if s['module'] in ('support','data','commercial') else 'accepted '+s['module']+' subcase', 'projection':'Internal caseId binds to the owning task case. Client caseId stays the parent request; the server resolves its scoped outstanding question/result, never an arbitrary child id.', 'join':'The Support root retains the customer objective and joins required child results and restoration obligations before delivery.'} record['case_scope']=journey['ui']['scope_note'] record['effect_requirements']={'derive_from':'Current capability, target audience, exact impact set and preflight; the displayed gate is one checkpoint, not the complete permission set.', 'rules':['Scoped transactional/compensable working action: HW-instance with approved shape.', 'Shared target: add HW-shared. DDL: add HW-ddl. Irreversible: add HW-irreversible.', 'Customer-facing environment: add H5. Customer-visible send: add H5-send.', 'H7 compensation retains all applicable effect prerequisites.'], 'example':'Working Configuration writes are scoped working unless this case identifies a shared target; final release/Data application examples use a customer-facing target.'} record['required_result']=journey['ui']['outcome'] record['stage_task']='Work only '+s['module']+' / '+str(s['stage'] or 'post-result')+'. '+s['done']+' Requested outcome: '+journey['ui']['outcome']+' Case task: '+record['narrative']['agent_message'] record['input_contracts']=s['input_contracts'];record['output_contracts']=s['output_contracts'] record['model_or_tool_work']=case['learn'] if protocol=='learning' else case['verify'] if protocol in ('verification','preview','delivery','test') else case['execute'] if protocol in ('execution','implementation','release','send') else case['investigate'] if protocol=='learning': record['http']+=copy.deepcopy(common['publication']['http']);record['storage']+=common['publication']['storage'] if protocol=='external': record['initialization']='Assigned external actor: '+str(journey['ui']['external_owner'])+'. Reported completion queues verification; it is not acceptance or a platform effect.' record['storage']=[x for x in record['storage'] if not x.startswith('Platform branch')] elif protocol=='execution': record= {**record,'http':copy.deepcopy(common['decision']['http'])+[x for x in record['http'] if x['method']=='GET']} record['storage']=[x for x in record['storage'] if not x.startswith('External branch')] record['initialization']='After the current effect gate, dispatch the exact journalled operation. No customer report endpoint is needed for a platform-owned action.' if protocol in ('scope','preview','delivery','offer_acceptance'): scope='offer' if protocol=='offer_acceptance' else 'scope' if protocol=='scope' else 'preview' if protocol=='preview' else 'delivery' for call in record['http']: if 'body' in call and 'scope' in call['body']:call['body']['scope']=scope record['initialization']='Business confirmation for '+scope+' only. It does not grant target write authority or complete a different scope.' if protocol=='implementation': record['storage']+=['PAPERS/build_state: case-owned worktree diff, source hashes and required tests. Publication/deployment are later effects.'] record['initialization']='Initialise the selected repository implementer after S2 placement approval. worktree.edit/diff/build stay confined; this is not deployment.' if protocol=='test': record['initialization']='source.test_runner executes the declared tests without a model prompt; source.reviewer receives their evidence at S5.' if protocol=='external_report': record['initialization']='Persist the attributed external action receipt. Schedule the required S6 target checks; a report is not a verified result and creates no platform WRITE_LOG effect.' record['storage']=['PAPERS: external response and actor/source evidence bound to the exact request, step and plan revision.','TASKS + OUTBOX_MESSAGES: resume the required verification with current evidence; no automatic acceptance.'] if protocol=='application_result': record['initialization']='Display the journalled operation disposition and actual effect receipt. S6 performs the independent customer-result checks before resolution.' if protocol=='core_request': record['initialization']='Communicator prepares the shared-core request, offer and proposed delivery terms under the Support root. Missing owner terms remain pending; no Source stage or implementation task is created.' if protocol=='referral': record['initialization']='Assess scope and record the programme owner. Do not instantiate S2–S8 as if the programme were a small change.' if protocol=='handover': record['initialization']='The receiving module root accepts the exact handover packet at Hd before its tasks start. The requesting root retains its outcome and waits for the signed child result.' if protocol=='stage_result': record['initialization']='H3 reviews only this stage sign-out. Its H2 plan and any working-target effect gates preceded execution; this is not target delivery or customer acceptance.' if protocol=='release': record['initialization']='After the release-plan gate, the target gate binds the exact release packet. The deterministic coordinator dispatches eligible components and records target verification; approval alone is not a completed release.' record['storage']+=copy.deepcopy(common['execution']['storage'][1:])+copy.deepcopy(common['verification']['storage']) # The two product surfaces never receive the same unfiltered endpoint/action set. customer=[{'method':'GET','path':'/client/requests/{caseId}','response':'200; customer-scoped request projection'}] internal=copy.deepcopy([x for x in record['http'] if not (x['method']=='GET' and x['path'].startswith('/client/'))]) if protocol in ('scope','preview','delivery','offer_acceptance'): customer+=copy.deepcopy([x for x in record['http'] if x['method']=='POST']) internal=[x for x in internal if x['method']=='GET'] elif protocol=='external': reports=[x for x in record['http'] if x['method']=='POST'] if journey['ui']['external_owner'] in ('Customer','Authorised customer operator'):customer+=copy.deepcopy(reports) for call in internal: if call['path']=='/client/requests/{caseId}/responses':call['path']='/cases/{caseId}/responses' elif protocol=='intake': record['ingress_requests']=copy.deepcopy([x for x in record['http'] if x['method']=='POST']) internal=[x for x in internal if x['method']=='GET'] record['surface_requests']={'internal':internal,'customer':customer} # Every example is tied to its stage. It cannot accidentally reuse another gate's command identity. for i,call in enumerate(record['http']): if call['method'] not in ('GET','HEAD'): call['headers']={'Idempotency-Key':case['id']+'-'+s['uid'].replace('/','-')+'-'+str(i)} if 'body' in call: body=call['body'] if 'gate_id' in body:body['gate_id']='fixture-'+s['uid'].replace('/','-')+'-'+str(s['active_gate']) if 'step_id' in body:body['step_id']=s['uid'] if 'artefact_hash' in body:body['artefact_hash']=hashlib.sha256((case['id']+s['uid']).encode()).hexdigest() for surface,calls in record['surface_requests'].items(): for i,call in enumerate(calls): if call['method'] not in ('GET','HEAD'): call['headers']={'Idempotency-Key':case['id']+'-'+s['uid'].replace('/','-')+'-'+surface+'-'+str(i)} if 'body' in call: if 'step_id' in call['body']:call['body']['step_id']=s['uid'] if 'gate_id' in call['body']:call['body']['gate_id']='fixture-'+s['uid'].replace('/','-')+'-'+str(s['active_gate']) if 'artefact_hash' in call['body']:call['body']['artefact_hash']=hashlib.sha256((case['id']+s['uid']).encode()).hexdigest() wanted=set(s['profiles']) if protocol in ('execution','release','reversal','send'): wanted.update(('platform.write_auditor','platform.write_executor')) if protocol in ('verification','preview','stage_result'): wanted.add(verification_profile(s['module'],s['stage'])) if protocol in ('send','delivery'):wanted.add('support.communicator') record['agent_initializations']=[] for a in base['agent_initializations']: if a['profile_key'] not in wanted:continue suffix='\n[CURRENT MODULE STAGE TASK]\n'+record['stage_task']+'\nInputs: '+', '.join(s['input_contracts'])+'\nOutputs: '+', '.join(s['output_contracts'])+'\nCurrent checkpoint: '+str(s['active_gate'] or 'no pending human gate')+'. '+record['initialization']+'\nCase-specific work: '+record['model_or_tool_work']+'\nReport missing inputs to the immediate parent. Do not claim any other stage or the parent objective complete.' task_anchor=s.get('navigation_uid',s['uid']) persistent=a['profile_key'].endswith('.root') or a['profile_key'] in ('commercial.assessor','commercial.coordinator','source.deployer','configuration.deployer') init={'template_task_id':a['task_id'],'task_id':a['task_id'] if persistent else a['task_id']+'-'+task_anchor.replace('/','-'), 'parent_task_id':a['parent_task_id'],'profile_key':a['profile_key'],'stage_id':None if persistent else s['stage'], 'phase_stage_id':s['stage'], 'execution_kind':a['execution_kind'],'task_text':record['stage_task'], 'lifecycle':'Open or resume only when this phase and its dependencies are ready. Root, assessment and delivery responsibilities retain their task and case brief across checkpoints; this snapshot is not an eager spawn.', 'rendered_suffix':suffix if a['execution_kind']=='llm' else None, 'input_sha256':hashlib.sha256((a['rendered_input']+suffix).encode()).hexdigest() if a['execution_kind']=='llm' else None} record['agent_initializations'].append(init) owner=next(a for a in record['agent_initializations'] if a['profile_key']==s['profile']) record['chat_protocol']['body']['context']={'task_id':owner['task_id'],'plan_id':'plan-fixture-'+case['id'].lower(),'plan_revision':1} record['chat_protocol']['subplans']=[] if case['id']=='CF-05' and s['module']=='configuration' and s['stage']=='S2': for i,row in enumerate(journey['ui']['rows'][:2]): record['chat_protocol']['subplans'].append({'key':'template-'+str(i),'name':row['label'], 'task_id':owner['task_id']+'-template-'+str(i),'parent_task_id':owner['task_id'], 'plan_id':'plan-fixture-cf-05-template-'+str(i),'plan_revision':1, 'profile_key':s['profile'],'task':'Prepare and check only the pinned template for '+row['label']+'. Return the artefact and positive/negative checks.', 'handshake':['Child proposes Plan v1','Immediate parent returns PlanConfirmation for preparation','Child returns bounded result; parent joins the stage packet'], 'authority':'No target writes or direct human questions. Stage H2 and effect gates remain independent.'}) for init in record['agent_initializations']: if init['profile_key']=='platform.write_executor':init['parent_task_id']=owner['task_id'] record['communication']={'stage_owner_profiles':s['profiles'], 'parent':'commercial.coordinator' if s['module']=='commercial' and s['stage'] in ('O1','O2') else 'support.root' if s['module'] in ('support','data','commercial') else s['module']+'.root', 'task_handshake':'Parent sends bounded task; stage owner returns Plan; immediate parent confirms the exact revision before dependent work.', 'result_contracts':s['output_contracts'],'completion_test':s['done'], 'return_path':'Result or Question returns to the immediate parent. Child root signs its joined result to the requesting root; only the main root asks the human.', 'verification':'Independent domain validator receives the original objective, source/specification citations, current baselines, relevant evidence, assertions, dependencies and known uncertainties. Exclude author reasoning/transcript, not necessary facts; request missing evidence from the parent.', 'reuse':'The stage owner rechecks applicability of pinned memory/skill/tool versions. Known mechanical segments run deterministically; only unresolved judgement invokes the model.'} out[s['uid']]=record return out