Merge pull request 'Schedule bridges on demand with durable fair message delivery' (#9) from codex/demand-message-scheduling into main

SKILL.md

@@ -132,6 +132,10 @@ description: 组织项目的多个 AI 会话并行工作，将进度、证据和
 
 ## 可靠完成上报与通知背压
 
+咨询桥接按真实待发或在途消息分配，不因绑定存在就长期占位。空闲释放前核对worker、待交付回执和未知执行；只停止自有桥接，不终止模型writer。保留并发上限，按持久等待时间公平轮转；人类咨询及决策优先，普通事项达到老化窗口后按先到顺序提升，不能永久饥饿。断网时保留原消息、幂等键和回执，恢复后续传；可能已执行的输入先查原会话和关联回复，禁止换ID盲重放。具体调度与恢复见运行指南。
+
+页面按证据区分已保存、排队、宿主送达、真实处理、关联回复和受阻。保存成功不等于送达，模型回复不等于业务验收；未知结果明确待核对。普通成功阶段收起保留可查，等待和受阻短提示，不堆叠技术回执。
+
 执行负责人完成工作后，先把带稳定报告ID、任务版本、证据和检查结果的完成报告保存到私有持久 outbox，再按有界退避重试交付；服务落库回执确认后才标已上报。重试复用同一幂等ID，不重新执行业务操作；结果未知先对账，达到上限保留待核查状态，交副主控处理。对话中的“完成”不能替代可靠落库，执行完成也不等于集成验收完成；集成负责人另核对源产物、接口和验收证据。
 
 通知按目标背压：同一目标最多一条未处理通知，后续事件保留在持久队列并按关联ID合并。排队accepted不等于处理完成，需核对真实目标turn/处理回执才能释放名额；未知结果不得通过换ID或清本地记录绕过。应用outbox、已知inflight与宿主原生队列分别记账；无法读取宿主总数或确认取消时，必须写未知/不支持，不能宣称原生队列已清空。现有机制与限制见运行指南。

agent_control/conversation.py

@@ -21,6 +21,27 @@ def browser_body(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}
@@ -101,5 +122,6 @@ def project(rows, links):
         if key in seen:
             continue
         seen.add(key)
-        result.append({**{k: row.get(k) for k in ('id', 'source', 'status', 'created')}, 'text': text, 'log_id':incident_links.get(group)})
+        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

agent_control/store.py

@@ -888,11 +888,13 @@ class Store:
 
     @staticmethod
     def public_message(row):
+        from .conversation import delivery_fields
         result = {k: row[k] for k in ('id','session_id','text','source','reply_to','status','created','response_id')}
         result['outcome'] = row['status'] if row['status'] in ('completed','failed','uncertain') else None
         result['origin'] = row['origin']
         result['execution'] = 'started' if row['started'] else 'not_started'
         result['submission'] = json.loads(row['submission']) if row['submission'] else None
+        result.update(delivery_fields(dict(row)))
         return result
 
     def send(self, data):
@@ -1905,9 +1907,16 @@ class Store:
                 value=self._decision(row)
                 if value['session_id']:
                     session=self._get(db,'sessions',value['session_id'])
+                    rows=db.execute("SELECT * FROM messages WHERE session_id=? AND source='human' AND origin='coordination' AND status IN ('queued','claimed','uncertain')", (value['session_id'],)).fetchall()
+                    now=time.time()
+                    queued=[m for m in rows if m['status']=='queued' or (m['status']=='claimed' and not m['started'] and (m['lease_until'] or 0)<=now)]
+                    uncertain=[m for m in rows if m['status']=='uncertain' or (m['status']=='claimed' and m['started'] and (m['lease_until'] or 0)<=now)]
+                    inflight=[m for m in rows if m not in queued and m not in uncertain]
+                    demand=dict(queued=len(queued),inflight=len(inflight),uncertain=len(uncertain),
+                                human_queued=len(queued),oldest_queued_at=min((m['created'] for m in queued),default=None))
                     bindings.append(dict(decision_id=value['id'],session_id=value['session_id'],thread_id=value['session_thread_id'],
                                          worktree=session['worktree'],status=value['status'],archive_status=value['archive_status'],
-                                         pending_messages=self._decision_pending(db,value)))
+                                         pending_messages=self._decision_pending(db,value),demand=demand))
             return {'bindings':bindings}
 
     def decision_consultation(self, did, data=None):

assets/dashboard/src/components/decision-chat.tsx

@@ -1,3 +1,4 @@
+import { MessageDelivery } from '@/components/message-delivery'
 import { useEffect, useRef, useState } from 'react'
 import { Button } from '@/components/ui/button'
 import { Textarea } from '@/components/ui/textarea'
@@ -44,7 +45,7 @@ export default function DecisionChat({ decision, token }: { decision: Decision;
   }
   return <section className="flex min-w-0 flex-col rounded-xl border bg-card" aria-label="本决策咨询聊天"><div className="flex items-center justify-between border-b p-3"><h3 className="font-medium">本决策咨询</h3><Button size="sm" variant="ghost" disabled={!bound} onClick={() => void refresh.current()}>刷新</Button></div>
     {(!bound || readOnly) && <p className="px-3 pt-3 text-xs text-muted-foreground">{!bound ? '咨询暂不可用。' : '已完成，只读。'}</p>}
-    <div ref={scroll} role="log" aria-label="本决策消息" className="max-h-[45vh] min-h-40 space-y-3 overflow-y-auto p-3">{messages.map(message => <article key={message.id} className={`rounded-lg p-3 text-sm ${message.source === 'human' ? 'ml-6 bg-primary text-primary-foreground' : 'mr-6 bg-muted'}`}><p className="mb-1 text-xs opacity-70">{message.source === 'human' ? '你' : '咨询助手'}</p><p className="whitespace-pre-wrap break-words">{message.text}</p>{message.source === 'human' && ['queued','claimed'].includes(message.status) && <p className="mt-1 text-xs">等待回复…</p>}{['failed','uncertain'].includes(message.status) && <p className="mt-1 text-xs">回复尚未确认，请勿重复发送。</p>}</article>)}{!messages.length && <p className="text-sm text-muted-foreground">{!bound ? '暂无已接通的咨询会话。' : loaded ? '暂无咨询消息。' : '正在读取咨询历史…'}</p>}</div>
+    <div ref={scroll} role="log" aria-label="本决策消息" className="max-h-[45vh] min-h-40 space-y-3 overflow-y-auto p-3">{messages.map(message => <article key={message.id} className={`rounded-lg p-3 text-sm ${message.source === 'human' ? 'ml-6 bg-primary text-primary-foreground' : 'mr-6 bg-muted'}`}><p className="mb-1 text-xs opacity-70">{message.source === 'human' ? '你' : '咨询助手'}</p><p className="whitespace-pre-wrap break-words">{message.text}</p><MessageDelivery message={message} messages={messages} /></article>)}{!messages.length && <p className="text-sm text-muted-foreground">{!bound ? '暂无已接通的咨询会话。' : loaded ? '暂无咨询消息。' : '正在读取咨询历史…'}</p>}</div>
     {(error || draft.error) && <p role="status" className="px-3 pb-3 text-sm text-destructive">{draft.error || error}</p>}
     <div className="space-y-2 px-3 pb-3">{local.error && <p role="status" className="text-xs text-destructive">{local.error}</p>}{local.error && <Button variant="outline" size="sm" disabled={draft.sending} onClick={() => persist()}>重试保存草稿</Button>}{draft.revision !== revision && <div className="text-sm"><p>这是版本 {draft.revision} 的咨询草稿，请重新确认当前问题。</p>{!readOnly && <Button variant="outline" size="sm" onClick={() => {draft.revision=revision;draft.attempt=undefined;persist()}}>确认用于当前版本</Button>}</div>}{readOnly && draft.text && <p className="whitespace-pre-wrap text-sm">保留的未发送草稿（只读）：{draft.text}</p>}</div>
     {!readOnly && <form className="space-y-2 border-t p-3" onSubmit={event => { event.preventDefault(); void send() }}><Textarea aria-label="本决策咨询问题" value={draft.text} disabled={!bound || draft.sending} maxLength={32000} onChange={event => { draft.text = event.target.value; persist() }} placeholder="询问方案、依据或影响…" /><Button type="submit" disabled={!bound || !local.ready || draft.revision !== revision || draft.sending || !draft.text.trim()}>{draft.sending ? '正在发送…' : '发送咨询'}</Button></form>}

assets/dashboard/src/components/message-delivery.tsx

@@ -0,0 +1,10 @@
+import type { Message } from '@/lib/types'
+import { messageDelivery } from '@/lib/message-delivery'
+export function MessageDelivery({ message, messages }: { message: Message; messages: Message[] }) {
+  if (message.source !== 'human') return null
+  const delivery = messageDelivery(message, messages)
+  return <details className="mt-2 text-xs opacity-80" open={delivery.waiting || undefined}>
+    <summary className="cursor-pointer select-none">{delivery.label}</summary>
+    <p className="mt-1 leading-5">{delivery.reason || (delivery.stage === 'replied' ? '已收到此消息的回复。' : delivery.stage === 'delivered' ? '已进入会话，等待处理与回复。' : delivery.stage === 'processing' ? '会话正在处理，尚未收到回复。' : delivery.stage === 'queued' ? '已在服务端保存，等待会话处理。' : '已保存，尚未确认送达。')}</p>
+  </details>
+}

assets/dashboard/src/lib/message-delivery.ts

@@ -0,0 +1,17 @@
+import type { Message } from './types'
+export type DeliveryStage = 'saved' | 'queued' | 'delivered' | 'processing' | 'replied' | 'blocked' | 'unknown'
+export const deliveryLabels: Record<DeliveryStage, string> = { saved:'已保存', queued:'排队中', delivered:'已送达', processing:'处理中', replied:'已回复', blocked:'受阻', unknown:'待核查' }
+export function messageDelivery(message: Message, messages: Message[]) {
+  const replied = messages.some(row => row.source === 'agent' && row.reply_to === message.id && row.session_id === message.session_id)
+  let stage: DeliveryStage
+  if (replied) stage = 'replied'
+  else if (message.delivery_stage && message.delivery_stage in deliveryLabels) stage = message.delivery_stage
+  else if (['failed','uncertain','blocked'].includes(message.status)) stage = 'blocked'
+  else if (message.status === 'queued') stage = 'queued'
+  else if (message.status === 'delivered' || message.submission?.state === 'accepted') stage = 'delivered'
+  else if (message.status === 'claimed') stage = 'queued' // A lease alone is not a running native turn.
+  else if (message.status === 'saved') stage = 'saved'
+  else stage = 'unknown' // Transport completion alone does not prove a visible reply.
+  const reason = stage === 'blocked' ? (message.delivery_reason || '结果尚未确认，正在核查；请勿重复发送。') : stage === 'unknown' ? '暂无法确认消息阶段，请刷新查看。' : ''
+  return {stage, label:deliveryLabels[stage], reason, waiting:['queued','processing','blocked','unknown'].includes(stage)}
+}

assets/dashboard/src/lib/types.ts

@@ -8,7 +8,7 @@ export type Task = {
  integration_status?: 'not_started' | 'waiting' | 'integrating' | 'integrated' | 'verified'; id: string; name: string; status: string; phase: string; eta_min_minutes?: number | null; eta_max_minutes?: number | null; eta_basis?: string; eta_estimated_at?: number | null; eta_revision?: number; eta_change_reason?: string; eta_last_reviewed_at?: number | null; eta_review_status?: string; current_progress?: string; progress_updated?: number | null; decision?: string; updated?: number }
 export type Session = { id: string; name: string; wp?: string; role?: string; thread_id?: string; thread_url?: string; connection?: string; history_synced_at?: number }
 export type Event = { id: number; kind: string; text: string; created: number; session_id?: string }
-export type Message = { log_id?: string | null; id: string; session_id: string; text: string; source: 'human' | 'agent'; status: string; created: number; origin?: string; reply_to?: string }
+export type Message = { delivery_stage?: 'saved' | 'queued' | 'delivered' | 'processing' | 'replied' | 'blocked'; delivery_reason?: string; submission?: {state?: string}; log_id?: string | null; id: string; session_id: string; text: string; source: 'human' | 'agent'; status: string; created: number; origin?: string; reply_to?: string }
 export type Sample<T> = { host?: string; status?: string; error?: string; collected_at?: string; data?: T | null }
 export type NodeMetric = { memory_used_percent?: number; memory_total_bytes?: number; memory_available_bytes?: number; load_1m?: number; load_5m?: number; load_15m?: number }
 export type Mount = { mount_target: string; used_percent?: number; total_bytes?: number; available_bytes?: number }

assets/dashboard/src/pages/communication.tsx

@@ -1,3 +1,4 @@
+import { MessageDelivery } from '@/components/message-delivery'
 import { useEffect, useRef, useState } from 'react'
 import type { FormEvent } from 'react'
 import { Link } from 'react-router-dom'
@@ -83,14 +84,13 @@ export default function Communication() {
             <p className="mb-1 text-xs opacity-65">{m.source === 'human' ? '你' : '项目助手'}</p>
             <p className="whitespace-pre-wrap wrap-anywhere text-sm leading-7">{m.text}</p>
             {m.log_id && <Link className="mt-2 inline-block text-sm text-primary underline underline-offset-4" to={`/logs?incident=${encodeURIComponent(m.log_id)}`}>查看日志</Link>}
-            {m.source === 'human' && ['queued', 'claimed'].includes(m.status) && <p className="mt-2 text-xs opacity-70">正在等待回复…</p>}
+            <MessageDelivery message={m} messages={items} />
-            {m.source === 'human' && ['failed', 'uncertain'].includes(m.status) && <p className="mt-2 text-xs opacity-70">暂未确认回复，请稍后查看；无需重复发送。</p>}
           </div>
         </article>) : <div className="flex h-full items-center justify-center text-sm text-muted-foreground">{known ? '有什么想了解的？从这里开始交流。' : '会话准备好后，即可在这里交流。'}</div>}
       </div>
       <form onSubmit={send} className="shrink-0 border-t p-3 sm:px-8">
         {local.error && <p role="status" className="mb-2 text-xs text-destructive">{local.error}</p>}<Textarea id="message" aria-label="输入消息" value={text} disabled={sending} onChange={e => local.save({...local.value,text:e.target.value})} placeholder="输入你的问题或想法…" maxLength={32000} rows={2} required className="resize-none border-0 shadow-none focus-visible:ring-0" />
-        <div className="mt-2 flex items-center justify-between gap-3"><p className="text-xs text-muted-foreground">{!known ? '正在准备会话' : session?.connection === 'connected' ? '' : '暂时离线，消息会保存并等待回复。'}</p><Button type="submit" disabled={!known || !local.ready || sending || !text.trim()}><Send />{sending ? '发送中…' : '发送'}</Button></div>
+        <div className="mt-2 flex items-center justify-between gap-3"><p className="text-xs text-muted-foreground">{!known ? '正在准备会话' : session?.connection === 'connected' ? '' : '会话暂时离线；已提交的消息等待接通。'}</p><Button type="submit" disabled={!known || !local.ready || sending || !text.trim()}><Send />{sending ? '发送中…' : '发送'}</Button></div>
       </form>
     </section>
   </div>

assets/dashboard/tests/message-delivery.test.mjs

@@ -0,0 +1,9 @@
+import test from 'node:test'
+import assert from 'node:assert/strict'
+import { messageDelivery } from '../src/lib/message-delivery.ts'
+const m={id:'m',session_id:'s',source:'human',text:'question',status:'queued'}
+test('each explicit transport stage remains separate',()=>{for(const stage of ['saved','queued','delivered','processing','replied','blocked'])assert.equal(messageDelivery({...m,delivery_stage:stage},[]).stage,stage)})
+test('receipt accepted proves delivery but never processing or reply',()=>{assert.equal(messageDelivery({...m,status:'claimed',submission:{state:'accepted'}},[]).stage,'delivered');assert.equal(messageDelivery({...m,status:'completed'},[]).stage,'unknown')})
+test('only real correlated same-session reply closes pending state',()=>{const reply={...m,id:'a',source:'agent',reply_to:'m'};assert.equal(messageDelivery(m,[reply]).stage,'replied');assert.equal(messageDelivery(m,[{...reply,session_id:'other'}]).stage,'queued');assert.equal(messageDelivery(m,[{...reply,reply_to:'other'}]).stage,'queued')})
+test('unknown failed and uncertain outcomes never invite replay',()=>{for(const status of ['failed','uncertain']){const result=messageDelivery({...m,status},[]);assert.equal(result.stage,'blocked');assert.match(result.reason,/请勿重复发送/)}assert.equal(messageDelivery({...m,status:'future'},[]).stage,'unknown')})
+test('delayed blocked receipt yields to a correlated reply',()=>{assert.equal(messageDelivery({...m,status:'uncertain',delivery_stage:'blocked'},[{...m,id:'a',source:'agent',reply_to:'m'}]).stage,'replied')})

references/design-system.md

@@ -35,7 +35,7 @@
 
 只显示用户实际输入、面向用户的最终回复及需要行动的简洁提醒。隐藏角色/安全提示封装、JSON事实快照、内部跨会话原始汇报、工具日志和合成验收整轮内容；原始消息、回执、来源和审计不删除。用户自己输入或引用的正文不能因包含“日志/测试/JSON”等关键词而过滤。
 
-页面固定连接人类入口，不显示会话选择、绑定状态说明、原会话跳转或新建会话表单。页头只提供“刷新”，重新读取聊天和必要状态，保留登录、草稿、目标与幂等发送记录，不做浏览器整页重载。正常聊天不堆时间、来源、投递徽标；只在等待或失败时用自然语言提示。
+页面固定连接人类入口，不显示会话选择、绑定状态说明、原会话跳转或新建会话表单。页头只提供“刷新”，重新读取聊天和必要状态，保留登录、草稿、目标与幂等发送记录，不做浏览器整页重载。正常聊天不堆时间、来源、投递徽标；每条实际网页输入的投递阶段用简短可展开文字查询，普通成功阶段默认收起，排队、处理中及受阻直接提示。阶段依据真实API投影，已保存不冒充送达，已回复不代表业务完成；不显示私有宿主回执或内部队列载荷。
 
 聊天读取独立GET /api/conversation投影视图；原始GET /api/messages继续用于后台审计、桥接和对账。投影通过保存的turn/phase/reply关系隐藏内部或验收轮次；网页封装只有与已保存原输入完整匹配才折叠为原文。重大真实故障显示人类可读摘要，不透出原始事件载荷。首次打开显示最近对话，阅读旧消息时不强制滚到底部。
 

references/operations.md

@@ -38,6 +38,16 @@ HTTP服务适合受控局域网或受保护隧道。跨不可信网络使用经
 
 ## 当前会话与桥接
 
+### 按需调度与断网续传
+
+决策bindings中的`demand`汇总真实持久队列：`queued/inflight/uncertain/human_queued/oldest_queued_at`，不依赖最近500条历史截断。过期且未开始的领取可重新领取；已开始过期或未知的输入只允许核对，不能自动重放。`pending_messages`保留兼容。
+
+fleet按需启动绑定会话的worker/mirror，空闲后安全释放桥接名额，保留原状态文件和真实会话。排队顺序与等待年龄持久保存；人类咨询优先且有界老化，已有在途执行不被抢占。未知执行保留镜像核对能力，不通过杀模型、删除状态或扩大并发解决等待。主机不可用/配置不符时保留受阻信息，不能伪称已处理。
+
+通知器优先决策，其他通知首次进入待发集合的时间持久保存，等待120秒后按先到顺序提升；服务按最早未确认顺序分页返回，断网重启不清等待年龄。提升不绕过每目标背压，已发送结果未知仍以原线程、完整输入与回执对账。镜像确认丢失时用相同外部item身份重传，服务端幂等合并，不重新执行模型。
+
+消息投影`delivery_stage`为`queued/delivered/processing/replied/blocked`；客户端保存确认可短暂显示`saved`。`accepted`宿主排队回执才显示送达，观察到真实turn才显示处理中，关联完成回复才显示已回复；本地started/submitting不能冒充送达。投影不公开完整submission或私有回执。界面不能仅凭已回复判断任务验收完成。
+
 加载skill时默认绑定当前真实会话；用户明确要求专门入口才另建，并将其设为默认。用宿主受支持工具确认真实ID，建立项目归属与看板映射。不要把内部执行器thread和桌面会话混为一谈。
 
 执行器使用既有模型登录。集群未认证时让本机运行worker和mirror，服务/指标留在集群；模型凭据不离开本机。远程API可用 `control.transport="ssh"`：配置入口账号、独立私钥、`remote_client` 与 `remote_config`。客户端使用严格主机校验，在服务端内存读取服务令牌；不把令牌或模型认证拷进前端。

scripts/decision_fleet.py

@@ -19,6 +19,28 @@ def busy(path):
     return bool(state.get('inflight') or state.get('pending_completion') or state.get('needs_reconciliation'))
 
 
+def reconcile_completed(path,client,session_id,thread_id):
+    """Clear only a proven terminal original execution, never submit another input."""
+    state=json.loads(path.read_text(encoding='utf-8'))
+    record=state.get('inflight') or {}
+    proof=record.get('submission') or {}
+    if not record.get('message_id') or not proof.get('input_sha256'):return False
+    from urllib.parse import quote
+    rows=client.call('/api/messages?session_id='+quote(session_id,safe=''))['messages']
+    original=next((m for m in rows if m['id']==record['message_id']),None)
+    if not original:return False
+    reply=next((m for m in rows if m['id']==original.get('response_id')),None)
+    remote=original.get('submission') or {}
+    if (not reply or original.get('status') not in ('completed','failed')
+            or remote.get('thread_id')!=thread_id
+            or remote.get('input_sha256')!=proof['input_sha256']
+            or (proof.get('turn_id') and remote.get('turn_id')!=proof['turn_id'])):return False
+    state.setdefault('delivery_receipts',{})[record['message_id']]=remote
+    for key in ('inflight','pending_completion','needs_reconciliation'):state.pop(key,None)
+    atomic_json(path,state)
+    return True
+
+
 def child_config(base,binding,root):
     did=binding['decision_id'];sid=binding['session_id'];thread=binding['thread_id']
     if any(not isinstance(v,str) or not re.fullmatch(r'[A-Za-z0-9_.:-]{1,128}',v) for v in (did,sid)):
@@ -54,20 +76,20 @@ class Fleet:
 
     def stop(self,did,force=False):
         child=self.children[did]
-        if not force and busy(Path(child['cfg']['state_file'])):return False
+        # Single-cycle children exit naturally; never interrupt a native writer.
-        for process in child['processes'].values():
+        if any(p.poll() is None for p in child['processes'].values()):return False
-            if process.poll() is None:process.terminate()
-        for process in child['processes'].values():
-            try:process.wait(timeout=10)
-            except subprocess.TimeoutExpired:process.kill();process.wait(timeout=10)
         for stream in child['logs'].values():stream.close()
         del self.children[did]
         return True
 
     def tick(self,bindings):
-        live={}
+        now=time.time();problems=[];live={};threads=set();sessions=set()
-        waiting=0;problems=[]
+        journal_path=self.root/'schedule.json'
-        threads=set();sessions=set()
+        journal=json.loads(journal_path.read_text(encoding='utf-8')) if journal_path.exists() else {}
+        notifications={}
+        if self.cfg.get('notification_state_file'):
+            notification_path=Path(self.cfg['notification_state_file'])
+            if notification_path.exists():notifications=json.loads(notification_path.read_text(encoding='utf-8')).get('deliveries',{})
         for binding in bindings:
             did=binding['decision_id']
             if did in live or binding['thread_id'] in threads or binding['session_id'] in sessions:
@@ -75,43 +97,66 @@ class Fleet:
             live[did]=binding;threads.add(binding['thread_id']);sessions.add(binding['session_id'])
         for did in list(self.children):
             binding=live.get(did)
-            if not binding or binding.get('archive_status')=='archived':
+            child=self.children[did]
-                if not self.stop(did):problems.append('archive_requires_reconciliation:'+did)
+            mirror=child['processes'].get('mirror')
-        for did,binding in sorted(live.items()):
+            if mirror and mirror.poll()==0:
+                journal.setdefault(did,{}).setdefault('fact_synced',{}).update(child.get('terminal_facts',{}))
+            if binding and child_config(self.cfg,binding,self.root)[0]!=self.children[did]['cfg']:
+                problems.append('binding_changed:'+did);continue
+            if not self.stop(did) and (not binding or binding.get('archive_status')=='archived'):
+                problems.append('archive_requires_reconciliation:'+did)
+        candidates=[]
+        for did,binding in live.items():
             if binding.get('archive_status')=='archived':continue
             candidate,runtime=child_config(self.cfg,binding,self.root)
-            child=self.children.get(did)
+            demand=binding.get('demand')
-            if child and child['cfg']!=candidate:
-                problems.append('binding_changed:'+did);continue
-            if not child:
-                if len(self.children)>=self.limit:waiting+=1;continue
-                config_path=runtime/'bridge.json'
-                if config_path.exists() and json.loads(config_path.read_text())!=candidate:
-                    problems.append('saved_binding_changed:'+did);continue
-                atomic_json(config_path,candidate)
-                child={'cfg':candidate,'runtime':runtime,'path':config_path,'processes':{},'logs':{},'retry_at':{},'started':{}}
-                self.children[did]=child
             state_path=Path(candidate['state_file'])
-            for role in ('worker','mirror'):
+            local_busy=busy(state_path)
-                process=child['processes'].get(role)
+            worker=self.children.get(did,{}).get('processes',{}).get('worker')
-                if process and process.poll() is None:
+            worker_alive=worker is not None and worker.poll() is None
-                    try:
+            if local_busy and not worker_alive and demand is not None and not demand.get('uncertain',0) and not demand.get('inflight',0):
-                        health=json.loads((runtime/(role+'-health.json')).read_text())
+                try:
-                        healthy=health.get('status')=='ok' and time.time()-health.get('updated',0)<60
+                    if reconcile_completed(state_path,Client(self.cfg['control']),binding['session_id'],binding['thread_id']):local_busy=False
-                    except (OSError,ValueError):healthy=False
+                except Exception:pass
-                    if not healthy and time.time()-child['started'].get(role,0)>60:
+            if demand is None:
-                        problems.append('bridge_health_stale:'+did+':'+role)
+                problems.append('queue_summary_unavailable:'+did);demand={}
-                    continue
+            uncertain=bool(demand.get('uncertain',0) or local_busy)
-                if role=='worker' and busy(state_path):
+            facts={key:r for key,r in notifications.items() if r.get('decision_id')==did and r.get('event_kind')=='fact_answered'}
-                    problems.append('execution_requires_reconciliation:'+did);continue
+            terminal_facts={key:str(r.get('completed_at') or r.get('turn_id') or 'done') for key,r in facts.items() if r.get('processing_done')}
-                if time.monotonic()<child['retry_at'].get(role,0):continue
+            synced=journal.get(did,{}).get('fact_synced',{})
-                if role not in child['logs']:child['logs'][role]=open(runtime/(role+'.log'),'a',encoding='utf-8')
+            fact_demand=any(not r.get('processing_done') for r in facts.values()) or any(synced.get(key)!=value for key,value in terminal_facts.items())
-                command=[sys.executable,'-X','utf8',str(Path(__file__).with_name(role+'.py')),'--config',str(child['path'])]
+            needed=bool(demand.get('queued',0) or demand.get('inflight',0) or uncertain or fact_demand)
-                child['processes'][role]=subprocess.Popen(command,stdout=child['logs'][role],stderr=child['logs'][role],
+            if not needed:
+                if did in journal:journal[did].pop('waiting_since',None)
+                continue
+            record=journal.setdefault(did,{})
+            record.setdefault('waiting_since',min(now,demand.get('oldest_queued_at') or now))
+            if did in self.children:
+                if uncertain and not worker_alive:problems.append('execution_requires_reconciliation:'+did)
+                continue
+            # Finite human boost: old ordinary work eventually precedes new human work.
+            score=now-record['waiting_since']+(60 if demand.get('human_queued',0) else 0)
+            candidates.append((-score,record['waiting_since'],record.get('last_served',0),did,binding,candidate,runtime,uncertain,terminal_facts))
+        candidates.sort(key=lambda row:row[:4]);waiting=0
+        for _,_,_,did,binding,candidate,runtime,uncertain,terminal_facts in candidates:
+            if len(self.children)>=self.limit:waiting+=1;continue
+            config_path=runtime/'bridge.json'
+            if config_path.exists() and json.loads(config_path.read_text(encoding='utf-8'))!=candidate:
+                problems.append('saved_binding_changed:'+did);continue
+            atomic_json(config_path,candidate)
+            child={'cfg':candidate,'runtime':runtime,'path':config_path,'processes':{},'logs':{},'terminal_facts':terminal_facts}
+            self.children[did]=child
+            journal[did].update(waiting_since=now,last_served=now)
+            atomic_json(journal_path,journal)
+            roles=('mirror',) if uncertain or binding.get('demand',{}).get('inflight',0) or not binding.get('demand',{}).get('queued',0) else ('worker','mirror')
+            if uncertain:problems.append('execution_requires_reconciliation:'+did)
+            for role in roles:
+                stream=open(runtime/(role+'.log'),'a',encoding='utf-8');child['logs'][role]=stream
+                command=[sys.executable,'-X','utf8',str(Path(__file__).with_name(role+'.py')),'--config',str(config_path),'--once']
+                child['processes'][role]=subprocess.Popen(command,stdout=stream,stderr=stream,
                     **({'creationflags':subprocess.CREATE_NO_WINDOW} if os.name=='nt' else {}))
-                child['started'][role]=time.time()
+        atomic_json(journal_path,journal)
-                child['retry_at'][role]=time.monotonic()+15
+        snapshot={'status':'failed' if problems else 'ok','updated':now,'active':len(self.children),
-        snapshot={'status':'failed' if problems else 'ok','updated':time.time(),'active':len(self.children),
                   'waiting':waiting,'problems':problems,'bindings':{did:{'session_id':c['cfg']['session_id'],
                     'thread_id':c['cfg']['thread_id'],'processes':{r:{'pid':p.pid,'running':p.poll() is None}
                     for r,p in c['processes'].items()}} for did,c in self.children.items()}}
@@ -119,7 +164,10 @@ class Fleet:
         return snapshot
 
     def close(self):
-        for did in list(self.children):self.stop(did,force=True)
+        # Active cycles retain their durable state and host writer locks.
+        for did in list(self.children):
+            for stream in self.children[did]['logs'].values():stream.close()
+            self.stop(did)
 
 
 def main():

scripts/notifier.py

@@ -59,8 +59,26 @@ def notification_text(n):
             +json.dumps(data,ensure_ascii=False)+'\n请核实事件与影响，按既定权限处理。面向用户仅汇报持续重大影响、需行动或待决策；短暂自动恢复只进日志，不向人类入口转发。若有log_id，给一句摘要并链接日志页#/logs?incident=<log_id>，不复制原载荷；重复状态不打扰。消息已投递不代表故障已修复。')
 
 
-def target_ready(cfg,records,thread,cache):
+def schedule_notifications(notifications,state,now):
-    if not cfg.get('enforce_backpressure'):
+    """Prefer decisions while aging waiting work into FIFO; persist across restarts.
+
+    This changes dispatch order only, never authorization, receipt or event identity.
+    A busy/uncertain target remains blocked regardless of waiting time.
+    """
+    waiting=state.setdefault('notification_waiting',{})
+    def identity(n):
+        return ('decision:' if 'decision' in n else 'eta:' if 'checkpoint' in n else 'hook:')+n['notification_id']
+    for n in notifications:
+        waiting.setdefault(identity(n),now)
+    def rank(n):
+        first=waiting[identity(n)]
+        priority=0 if 'decision' in n else 1 if n.get('hook',{}).get('severity')=='critical' else 2 if 'hook' in n else 3
+        return (0 if now-first>=120 else priority+1,first,identity(n))
+    return sorted(notifications,key=rank)
+
+
+def target_ready(cfg,records,thread,cache,force_reconcile=False):
+    if not cfg.get('enforce_backpressure') and not force_reconcile:
         return True
     limit=cfg.get('max_inflight_per_target',1)
     if not isinstance(limit,int) or isinstance(limit,bool) or not 1<=limit<=3:
@@ -96,6 +114,15 @@ def deliver(client,cfg,state,path):
     records=state.setdefault('deliveries',{})
     errors=[]
     cache={}
+    # Fact answers enter the native consultation thread directly. Reconcile their
+    # completion even with an empty outbox, so a demand-driven mirror can retire.
+    for thread in {r['thread_id'] for r in records.values()
+                   if r.get('event_kind')=='fact_answered' and not r.get('processing_done')}:
+        try:
+            target_ready(cfg,records,thread,cache,force_reconcile=True)
+        except Exception:
+            errors.append('fact_answer_reconciliation_unavailable')
+    atomic_json(path,state)
     if cfg.get('coalesce_stale_hooks'):
         # Upgrade old accepted-but-unacked receipts before pruning pending data.
         # Unknown submissions are protected, never silently cancelled or replayed.
@@ -128,7 +155,9 @@ def deliver(client,cfg,state,path):
             notifications += batch['notifications']
             queue_counts[endpoint]=batch.get('pending_count',len(batch['notifications']))
         except Exception:errors.append('outbox_unavailable')
-    notifications.sort(key=lambda n:0 if 'decision' in n else 1 if n.get('hook',{}).get('severity')=='critical' else 2 if 'hook' in n else 3)
+    notifications=schedule_notifications(notifications,state,time.time())
+    # Waiting age must survive offline delivery/ACK failures, not just successful sends.
+    atomic_json(path,state)
     task_owners={}
     if cfg.get('execution_notification_targets'):
         snapshot=client.call('/api/state')
@@ -161,6 +190,8 @@ def deliver(client,cfg,state,path):
                 if 'decision' in n:
                     client.call('/api/decision-notifications/submitting',{'notification_id':n['notification_id']})
                 old={'kind':'hook' if 'hook' in n else 'decision' if 'decision' in n else 'eta','notification_ids':([item['notification_id'] for item in notifications if item.get('hook')==n.get('hook')] if internal_event else [n['notification_id']]),'created_at':time.time(),'thread_id':thread,'input_sha256':digest,'state':'submitting'}
+                if 'decision' in n:
+                    old.update(decision_id=n['decision']['id'],event_kind=n.get('event',{}).get('kind'))
                 records[key]=old
                 atomic_json(path,state)
                 if 'hook' in n and cfg.get('coalesce_stale_hooks'):
@@ -177,6 +208,7 @@ def deliver(client,cfg,state,path):
                 # Crash after queue acceptance: reconcile exact actual input, never requeue.
                 with AppServer(cfg['codex']) as reader:
                     meta,turns=reader.history(thread)
+                if meta.get('id')!=thread:raise ValueError('Unexpected reconciliation thread identity')
                 observed=next((t['id'] for t in turns if any(
                     i.get('type')=='userMessage' and '\n'.join(p.get('text','') for p in i.get('content',[]) if p.get('type')=='text')==prompt
                     for i in t.get('items',[]))),None)

tests/test_decision_bridge.py

@@ -25,7 +25,8 @@ class DecisionBridgeTests(unittest.TestCase):
         cfg={'control':{},'codex':'codex','state_file':str(Path(root)/'worker.json'),
              'decision_fleet':{'runtime_root':str(Path(root)/'decisions'),'enabled':True,'max_active':1}}
         binding={'decision_id':'d','session_id':'s','thread_id':'11111111-1111-4111-8111-111111111111',
-                 'worktree':root,'status':'open','archive_status':'none'}
+                 'worktree':root,'status':'open','archive_status':'none',
+                 'demand':{'queued':1,'human_queued':1,'inflight':0,'uncertain':0}}
         return cfg,binding
 
     def test_only_bound_threads_and_bounded_startup_are_bridged(self):
@@ -65,6 +66,7 @@ class DecisionBridgeTests(unittest.TestCase):
                 self.assertIn('d',fleet.children)
                 self.assertTrue(all(p.poll() is None for p in child['processes'].values()))
                 atomic_json(state,{})
+                for process in child['processes'].values():process.code=0
                 fleet.tick([{**b,'status':'completed','archive_status':'archived'}])
                 self.assertEqual(fleet.children,{})
                 self.assertTrue(state.exists())

tests/test_decision_fleet.py

@@ -0,0 +1,76 @@
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
+import decision_fleet
+from test_decision_bridge import Process
+import test_decision_bridge
+
+class DemandTests(unittest.TestCase):
+    def setup_fleet(self,root):return test_decision_bridge.DecisionBridgeTests().fixture(root)
+    def test_idle_bindings_start_nothing(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);b['demand']['queued']=0
+            f=decision_fleet.Fleet(cfg)
+            with patch.object(decision_fleet.subprocess,'Popen') as spawn:
+                self.assertEqual(f.tick([b])['active'],0);spawn.assert_not_called()
+    def test_one_message_cycle_releases_and_old_waiter_wins(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);f=decision_fleet.Fleet(cfg)
+            other={**b,'decision_id':'z','session_id':'z','thread_id':'22222222-2222-4222-8222-222222222222'}
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
+                with patch.object(decision_fleet.time,'time',return_value=100):f.tick([b,other])
+                self.assertIn('--once',spawn.call_args.args[0])
+                for p in f.children['d']['processes'].values():p.code=0
+                with patch.object(decision_fleet.time,'time',return_value=110):f.tick([b,other])
+                self.assertEqual(set(f.children),{'z'});f.close()
+    def test_old_ordinary_precedes_fresh_human_after_restart(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);b['demand']={**b['demand'],'human_queued':0,'oldest_queued_at':1}
+            f=decision_fleet.Fleet(cfg)
+            other={**b,'decision_id':'a','session_id':'a','thread_id':'22222222-2222-4222-8222-222222222222','demand':{'queued':1,'human_queued':1}}
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()):
+                with patch.object(decision_fleet.time,'time',return_value=200):f.tick([b,other])
+                self.assertEqual(set(f.children),{'d'});f.close()
+                saved=json.loads((f.root/'schedule.json').read_text());self.assertEqual(saved['a']['waiting_since'],200)
+    def test_uncertain_after_restart_only_mirrors(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);b['demand']['uncertain']=1
+            f=decision_fleet.Fleet(cfg)
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
+                result=f.tick([b]);self.assertEqual(set(f.children['d']['processes']),{'mirror'})
+                self.assertEqual(spawn.call_count,1);self.assertEqual(result['status'],'failed');f.close()
+    def test_fact_reply_waits_for_native_terminal_and_final_mirror(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);b['demand']['queued']=0
+            cfg['notification_state_file']=str(Path(root)/'notifier.json')
+            path=Path(cfg['notification_state_file'])
+            record={'decision_id':'d','event_kind':'fact_answered','processing_done':False}
+            decision_fleet.atomic_json(path,{'deliveries':{'receipt':record}})
+            f=decision_fleet.Fleet(cfg)
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
+                f.tick([b]);self.assertEqual(set(f.children['d']['processes']),{'mirror'})
+                f.children['d']['processes']['mirror'].code=0
+                record.update(processing_done=True,completed_at=123)
+                decision_fleet.atomic_json(path,{'deliveries':{'receipt':record}})
+                f.tick([b]);self.assertEqual(spawn.call_count,2)
+                f.children['d']['processes']['mirror'].code=0
+                self.assertEqual(f.tick([b])['active'],0)
+                self.assertEqual(f.tick([b])['active'],0);self.assertEqual(spawn.call_count,2)
+                f.close()
+
+    def test_terminal_proof_clears_local_without_replaying(self):
+        with tempfile.TemporaryDirectory() as root:
+            path=Path(root)/'state.json';proof={'thread_id':'t','input_sha256':'h','turn_id':'turn'}
+            decision_fleet.atomic_json(path,{'inflight':{'message_id':'original','submission':proof},'needs_reconciliation':True})
+            class Client:
+                def call(self,*args):return {'messages':[{'id':'original','status':'completed','response_id':'reply','submission':proof},{'id':'reply','text':'done'}]}
+            self.assertFalse(decision_fleet.reconcile_completed(path,Client(),'s','different-thread'))
+            self.assertTrue(decision_fleet.busy(path))
+            self.assertTrue(decision_fleet.reconcile_completed(path,Client(),'s','t'))
+            self.assertFalse(decision_fleet.busy(path));self.assertIn('original',json.loads(path.read_text())['delivery_receipts'])
+
+if __name__=='__main__':unittest.main()

tests/test_message_demand.py

@@ -0,0 +1,68 @@
+"""Transport evidence and demand survive storage reopen without resending."""
+import json
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from agent_control.store import Store
+from agent_control.conversation import project
+from test_decision_workflow import bind
+from test_decisions import definition
+
+
+class DemandTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp=tempfile.TemporaryDirectory()
+        self.path=Path(self.tmp.name)/'state.sqlite3'
+        self.store=Store(self.path)
+        self.store.decision(definition())
+        bind(self.store)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def demand(self):
+        return Store(self.path).decision_bindings()['bindings'][0]['demand']
+
+    def test_persisted_queue_and_expired_claims_have_distinct_demand(self):
+        m=self.store.send(dict(session_id='consult',text='Question',idempotency_key='one'))
+        self.assertEqual(m['delivery_stage'],'queued')
+        self.assertEqual(self.demand()['queued'],1)
+        self.assertEqual(self.demand()['human_queued'],1)
+        self.assertIsNotNone(self.demand()['oldest_queued_at'])
+        claim=self.store.claim(dict(session_id='consult',worker_id='worker',lease_seconds=10))['message']
+        self.assertEqual(self.demand()['inflight'],1)
+        with self.store.db() as db:
+            db.execute('UPDATE messages SET lease_until=? WHERE id=?',(time.time()-1,m['id']))
+        self.assertEqual(self.demand()['queued'],1)
+        with self.store.db() as db:
+            db.execute('UPDATE messages SET started=1 WHERE id=?',(m['id'],))
+        self.assertEqual(self.demand()['uncertain'],1)
+        self.assertEqual(self.demand()['queued'],0)
+        self.assertEqual(Store(self.path).send(dict(session_id='consult',text='Question',idempotency_key='one'))['id'],m['id'])
+        self.assertEqual(len(self.store.messages('consult')),1)
+
+    def test_safe_projection_uses_host_evidence_not_local_start(self):
+        m=self.store.send(dict(session_id='consult',text='Question',idempotency_key='one'))
+        stages=[('claimed',None,None,'queued'),
+                ('claimed',dict(state='submitting'),None,'queued'),
+                ('claimed',dict(state='accepted',receipt_id='private'),None,'delivered'),
+                ('claimed',dict(state='observed',turn_id='private'),None,'processing'),
+                ('uncertain',dict(state='accepted'),None,'blocked'),
+                ('completed',dict(state='observed'),'reply','replied')]
+        for status,submission,response,expected in stages:
+            with self.store.db() as db:
+                db.execute('UPDATE messages SET status=?,started=1,submission=?,response_id=? WHERE id=?',
+                           (status,json.dumps(submission) if submission else None,response,m['id']))
+            view=Store(self.path).conversation('consult')[0]
+            self.assertEqual(view['delivery_stage'],expected)
+            self.assertNotIn('submission',view)
+            self.assertNotIn('private',json.dumps(view))
+
+    def test_mirrored_history_does_not_fabricate_delivery_proof(self):
+        row=dict(id='history',text='Question',source='human',origin='session_history',status='delivered',created=1)
+        self.assertNotIn('delivery_stage',project([row],[])[0])
+
+
+if __name__=='__main__':
+    unittest.main()

tests/test_mirror.py

@@ -123,5 +123,33 @@ class MirrorTests(unittest.TestCase):
             client.instance='two'
             with self.assertRaisesRegex(RuntimeError,'another control'): sync(Server(),client,cfg)
 
+    def test_lost_import_ack_retries_same_external_identity_without_model_execution(self):
+        class Server:
+            def history(self,tid):
+                return {'id':tid},[{'id':'turn','status':'completed','items':[
+                    {'id':'reply','type':'agentMessage','phase':'final_answer','text':'result'}]}]
+        class Client:
+            calls=[]
+            imported={}
+            lose_ack=True
+            def call(self,path,data=None):
+                if path.startswith('/api/messages'):return {'messages':[]}
+                if path=='/api/state':return {'instance_id':'one'}
+                self.calls.append(data)
+                for item in data['items']:self.imported[item['id']]=item
+                if self.lose_ack:
+                    self.lose_ack=False
+                    raise OSError('server persisted import but response was lost')
+                return {'deferred':[]}
+        with tempfile.TemporaryDirectory() as root:
+            cfg={'thread_id':'thread','session_id':'session','control':{},'state_file':str(Path(root)/'worker.json')}
+            client=Client()
+            with self.assertRaises(OSError):sync(Server(),client,cfg)
+            self.assertFalse((Path(root)/'mirror-state.json').exists())
+            sync(Server(),client,cfg);sync(Server(),client,cfg)
+            self.assertEqual(len(client.calls),2)
+            self.assertEqual(client.calls[0],client.calls[1])
+            self.assertEqual(list(client.imported),['reply'])
+
 
 if __name__=='__main__':unittest.main()

tests/test_notifier.py

@@ -10,6 +10,53 @@ import notifier
 
 
 class NotifierTests(unittest.TestCase):
+    def test_decisions_prefer_fresh_work_but_waiting_items_survive_restart_and_age(self):
+        normal={'notification_id':'ordinary','checkpoint':'half','task':{'id':'work'}}
+        decision={'notification_id':'choice','decision':{'id':'decision'}}
+        state={}
+        self.assertEqual(notifier.schedule_notifications([normal,decision],state,100)[0],decision)
+        # Persist/reload and a continuing stream of fresh decisions cannot starve old work.
+        state=json.loads(json.dumps(state))
+        fresh={'notification_id':'new-choice','decision':{'id':'new'}}
+        self.assertEqual(notifier.schedule_notifications([fresh,normal],state,221)[0],normal)
+        self.assertEqual(state['notification_waiting']['eta:ordinary'],100)
+
+    def test_waiting_age_does_not_bypass_uncertain_target_backpressure(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,n,path=self.fixture(root);cfg['enforce_backpressure']=True
+            class Client:
+                def call(self,p,data=None):return {'notifications':[n]} if p.endswith('notifications') else {}
+            state={'notification_waiting':{'hook:note1':0}}
+            with patch.object(notifier,'target_ready',return_value=False),patch.object(notifier.subprocess,'run') as run:
+                notifier.deliver(Client(),cfg,state,path)
+                run.assert_not_called()
+            self.assertEqual(json.loads(path.read_text())['notification_waiting']['hook:note1'],0)
+
+    def test_uncertain_reconciliation_rejects_wrong_thread_even_with_exact_input(self):
+        import hashlib
+        with tempfile.TemporaryDirectory() as root:
+            cfg,n,path=self.fixture(root)
+            prompt=notifier.notification_text(n)
+            state={'deliveries':{'note1':{'thread_id':cfg['notification_targets']['control'],
+                'input_sha256':hashlib.sha256(prompt.encode()).hexdigest(),'state':'submitting'}}}
+            class Client:
+                acks=[]
+                def call(self,p,data=None):
+                    if p.endswith('notifications'):return {'notifications':[n]}
+                    self.acks.append(p);return {}
+            class Reader:
+                def __init__(self,*args):pass
+                def __enter__(self):return self
+                def __exit__(self,*args):pass
+                def history(self,thread):return {'id':'different-thread'},[{'id':'turn','items':[
+                    {'type':'userMessage','content':[{'type':'text','text':prompt}]}]}]
+            client=Client()
+            with patch.object(notifier,'AppServer',Reader),patch.object(notifier.subprocess,'run') as run:
+                with self.assertRaises(RuntimeError):notifier.deliver(client,cfg,state,path)
+                run.assert_not_called()
+            self.assertEqual(client.acks,[])
+            self.assertEqual(state['deliveries']['note1']['state'],'submitting')
+
     def fixture(self,root):
         cfg={'control':{},'notification_targets':{'control':'11111111-1111-4111-8111-111111111111'},'codex':'codex','cwd':root}
         n={'notification_id':'note1','target':'control','hook':{'source':'test','event_id':'incident','summary':'synthetic failure','severity':'warning','status':'firing'}}
@@ -165,13 +212,57 @@ class NotifierTests(unittest.TestCase):
                'event':{'kind':'fact_answered','request_id':'fact-one','answer':'The prerequisite is verified.'}}
             class Client:
                 def call(self,p,data=None):return {'notifications':[n]} if p=='/api/decision-notifications' else {'notifications':[]}
+            class Reader:
+                def __init__(self,*args):pass
+                def __enter__(self):return self
+                def __exit__(self,*args):pass
+                def call(self,method,params):
+                    return {'thread':{'id':thread}} if method=='thread/read' else {'data':[]}
             result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+thread+'.')
             state={}
-            with patch.object(notifier.subprocess,'run',return_value=result) as run:
+            with patch.object(notifier.subprocess,'run',return_value=result) as run,patch.object(notifier,'AppServer',Reader):
                 notifier.deliver(Client(),cfg,state,path);notifier.deliver(Client(),cfg,state,path)
                 self.assertEqual(run.call_count,1)
                 self.assertIn(thread,run.call_args.args[0])
                 self.assertNotIn(cfg['notification_targets']['control'],run.call_args.args[0])
+            record=state['deliveries']['decision:fact-answer']
+            self.assertEqual(record['decision_id'],'one')
+            self.assertEqual(record['event_kind'],'fact_answered')
+
+    def test_fact_answer_completion_reconciles_empty_outbox_without_resend(self):
+        import hashlib
+        with tempfile.TemporaryDirectory() as root:
+            cfg,_,path=self.fixture(root)
+            thread='44444444-4444-4444-8444-444444444444'
+            prompt='fact answer exact input'
+            record={'kind':'decision','decision_id':'one','event_kind':'fact_answered',
+                    'thread_id':thread,'input_sha256':hashlib.sha256(prompt.encode()).hexdigest(),
+                    'state':'accepted','acked':True}
+            state={'deliveries':{'decision:answer':record}}
+            class Reader:
+                body=prompt
+                identity=thread
+                def __init__(self,*args):pass
+                def __enter__(self):return self
+                def __exit__(self,*args):pass
+                def call(self,method,params):
+                    if method=='thread/read':return {'thread':{'id':self.identity}}
+                    return {'data':[{'id':'real-turn','status':'completed','completedAt':123,
+                        'items':[{'type':'userMessage','content':[{'type':'text','text':self.body}]},
+                                 {'type':'agentMessage','phase':'final_answer','text':'verified'}]}]}
+            client=SimpleNamespace(call=lambda *args:{'notifications':[]})
+            with patch.object(notifier,'AppServer',Reader),patch.object(notifier.subprocess,'run') as run:
+                Reader.body=prompt+' changed'
+                notifier.deliver(client,cfg,state,path)
+                self.assertFalse(record.get('processing_done'))
+                Reader.body=prompt;Reader.identity='wrong-thread'
+                with self.assertRaises(RuntimeError):notifier.deliver(client,cfg,state,path)
+                self.assertFalse(record.get('processing_done'))
+                Reader.identity=thread
+                notifier.deliver(client,cfg,state,path)
+                run.assert_not_called()
+            self.assertEqual(record['turn_id'],'real-turn')
+            self.assertTrue(json.loads(path.read_text())['deliveries']['decision:answer']['processing_done'])
 
     def test_review_notice_is_concise_and_submission_is_not_acceptance(self):
         n={'notification_id':'notice','decision':{'id':'one','title':'Plan','options':[{'id':'a','label':'Private option detail'}]},