AISA v2.0 / Technical documentation / case_ui_model.py
case_ui_model.py
Python · 368 lines · 34,194 bytes · wiki path 10 Architecture/ui/case_ui_model.py · download the raw file · cited from Case protocol
Same folder: agent-catalog.json · agent_catalog.py · agent_map_svg.py · build_agent_map.py · build_case_content_docs.py · build_case_journeys.py · build_case_traces.py · build_operations_examples.py · case-content.json · case-journeys.json · case-runtime-traces.json · case-ui-specs.json · case_content.py · frequent-operations.json · memory-capabilities.json · validate_agent_map.py · validate_case_content.py
"""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