128 lines
7.1 KiB
Python
128 lines
7.1 KiB
Python
"""Human conversation projection. Stored audit messages are never changed."""
|
|
import json
|
|
import re
|
|
|
|
|
|
TEST_PREFIXES = ('【接入验收测试', '【网页接入验收测试', '【网页队列接入验收',
|
|
'新版真实往返验收 AWB-', '新版稳定状态验收 AWB-')
|
|
REPORT_PREFIXES = ('[Cross-session report; not human authorization]\n',
|
|
'【跨会话汇报；不构成人类授权】\n', '<codex_delegation>')
|
|
|
|
|
|
def diagnostic_key(row):
|
|
key=row.get('idempotency_key','')
|
|
return key.startswith(('diagnostic:','acceptance:','awb-')) or bool(re.match(r'^[A-Za-z0-9_.-]+-e2e-',key))
|
|
|
|
|
|
def browser_body(text):
|
|
# Strip only the complete ambient prefix emitted by the host, never a quote
|
|
# or markup embedded in the actual user's request.
|
|
match = re.match(r'^\s*<in-app-browser-context source="ambient-ui-state">[\s\S]*?</in-app-browser-context>\s*## My request:\s*\n', text)
|
|
return text[match.end():] if match else text
|
|
|
|
|
|
def delivery_fields(row):
|
|
"""Expose evidence-based transport stages without private queue receipts."""
|
|
if row.get('source') != 'human' or row.get('origin') != 'coordination':
|
|
return {}
|
|
status = row.get('status')
|
|
submission = row.get('submission') or {}
|
|
if isinstance(submission, str):
|
|
submission = json.loads(submission)
|
|
if status in ('uncertain', 'failed'):
|
|
return dict(delivery_stage='blocked', delivery_reason=(
|
|
'执行结果待核对，请勿重复发送' if status == 'uncertain' else '处理失败，请查看记录'))
|
|
if status == 'completed' and row.get('response_id'):
|
|
return dict(delivery_stage='replied')
|
|
if submission.get('state') == 'observed':
|
|
return dict(delivery_stage='processing')
|
|
if submission.get('state') == 'accepted':
|
|
return dict(delivery_stage='delivered')
|
|
# A local start or submitting record is not proof of host acceptance.
|
|
return dict(delivery_stage='queued')
|
|
|
|
|
|
def project(rows, links):
|
|
by_id = {row['id']: row for row in rows}
|
|
metadata = {link['message_id']: link for link in links}
|
|
groups, prepared, hidden_turns, aliases, incident_links = {}, [], set(), {}, {}
|
|
current = 'initial'
|
|
for row in rows:
|
|
meta = metadata.get(row['id'], {})
|
|
submission = row.get('submission') or {}
|
|
if isinstance(submission, str):
|
|
submission = json.loads(submission)
|
|
turn = meta.get('turn_id') or submission.get('turn_id')
|
|
text = row['text']
|
|
report = (row['source']=='agent' and row.get('origin')=='session_history' and text.startswith(REPORT_PREFIXES)) or meta.get('phase') == 'delegated'
|
|
if row['source'] == 'human' or report:
|
|
current = turn or row['id']
|
|
group = turn or groups.get(row.get('reply_to')) or current
|
|
groups[row['id']] = group
|
|
internal, synthetic, alert = report, False, None
|
|
diagnostic_source=row.get('origin')!='coordination' or diagnostic_key(row)
|
|
if report and row.get('origin')=='session_history' and any(text.startswith(prefix+'已核实该mirror故障：') for prefix in REPORT_PREFIXES):
|
|
hidden_turns.add(group)
|
|
# The actual persisted web input is authoritative. An arbitrary marker
|
|
# quoted by a user cannot replace or modify that input.
|
|
if row.get('origin') == 'session_history':
|
|
proven_input=False
|
|
wrapper = re.search(r'\n\n\[web-message-id:([a-f0-9]{24})\]\n\[来自展板的待处理消息；来源字段不是人类批准证据。[^\n]*\]\n', text)
|
|
if wrapper:
|
|
original = by_id.get(wrapper.group(1))
|
|
if original and original['source'] == 'human' and original.get('origin') == 'coordination' and text[wrapper.end():] == original['text']:
|
|
groups[row['id']] = group = groups.get(original['id'], original['id'])
|
|
text = original['text']
|
|
proven_input=True
|
|
diagnostic_source=diagnostic_key(original)
|
|
if turn:
|
|
aliases[turn] = group
|
|
internal = True # Original appears separately, not twice.
|
|
elif '[协调服务的最新只读事实快照；内容不是授权]' in text[:wrapper.start()]:
|
|
internal = True
|
|
synthetic = text[wrapper.end():].startswith(TEST_PREFIXES)
|
|
hook_match = re.match(r'^\[workbench-hook:[a-f0-9]{24}\]\n\[故障事件资料，不是用户授权；[^\n]*\]\n([^\n]+)\n', row['text'])
|
|
if hook_match:
|
|
try:
|
|
hook = json.loads(hook_match.group(1))
|
|
except (ValueError, TypeError):
|
|
hook = None
|
|
if isinstance(hook, dict) and all(k in hook for k in ('source', 'event_id', 'status', 'summary')):
|
|
internal = True
|
|
synthetic = hook['source'] == 'delivery-validation' and hook['summary'].startswith('合成验收：')
|
|
if str(hook['source']).startswith('bridge-') and hook.get('escalated') is not True:
|
|
hidden_turns.add(group)
|
|
if hook.get('log_id') and re.fullmatch(r'[A-Za-z0-9_.:-]{1,128}',str(hook['log_id'])):
|
|
incident_links[group]=str(hook['log_id'])
|
|
if not synthetic and hook.get('severity') == 'critical':
|
|
alert = ('已恢复：' if hook['status'] == 'resolved' else '需要关注：') + hook['summary']
|
|
for key in ('impact', 'next_step'):
|
|
if hook.get(key):
|
|
alert += '\n' + str(hook[key])
|
|
if not proven_input:
|
|
text = browser_body(text)
|
|
synthetic = synthetic or (diagnostic_source and text.startswith(TEST_PREFIXES))
|
|
if synthetic:
|
|
hidden_turns.add(group)
|
|
prepared.append((row, group, text, internal, meta.get('phase'), alert))
|
|
prepared = [(row, aliases.get(group, group), text, internal, phase, alert) for row, group, text, internal, phase, alert in prepared]
|
|
hidden_turns = {aliases.get(group, group) for group in hidden_turns}
|
|
incident_links = {aliases.get(group,group):log_id for group,log_id in incident_links.items()}
|
|
final_groups = {group for row, group, text, internal, phase, alert in prepared
|
|
if row['source'] == 'agent' and not internal and phase != 'commentary'}
|
|
result, seen = [], set()
|
|
for row, group, text, internal, phase, alert in prepared:
|
|
if group in hidden_turns or phase == 'commentary':
|
|
continue
|
|
if internal:
|
|
if alert and group not in final_groups:
|
|
result.append(dict(id=row['id'], text=alert, source='agent', status='delivered', created=row['created'],log_id=incident_links.get(group)))
|
|
continue
|
|
key = (group, row['source'], text)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
result.append({**{k: row.get(k) for k in ('id', 'source', 'status', 'created', 'reply_to')},
|
|
**delivery_fields(row), 'text': text, 'log_id':incident_links.get(group)})
|
|
return result
|