提议采纳：逐票快速核验、原子保存与重复确认幂等（台账 078）

一、盛弘股份09:45采纳因整榜读取超过3秒失败，实测3MB响应5.481秒。
改用已发布计划的逐票接口，取数失败保留原准备内容，已知研究变化仍阻断。
恢复只核验原数量，确认不重问模型、不新记消费；页面不把连接失败当建议变化。

二、普通建仓的采纳状态、动作审计和指令同事务提交。
API至真实SQL的13项离线反例覆盖故障回滚、双击、响应丢失重试和失效边界。
重复确认返回原指令，提议身份进入指令幂等键，避免同日不同建议碰撞。

三、47批隔离禁网回归和三页面守卫通过。候选镜像与155部署另记。
交易时段、排序、仓位、择时与下发终检没有改变。卖出与策略发单不并入本次事务改造。

DEVLOG.md

@@ -2088,3 +2088,15 @@ QMT在线、对账无差异。午休任务按原时段跳过，不制造交易
 恢复与异常细节见四仓交接11:40节。
 回退标签pms:rollback-before-prepared-20261020。
 回退须源码反向提交后拉取并重建。
+
+
+## 2026-10-21 采纳重请求与落库中断修复（开发验证）
+
+09:45盛弘股份采纳超时来自3秒整榜读取。桥现场5.481秒、3093481字节。
+新增逐票发布结果接口，采纳不下载整榜。网络暂不可用保留原准备内容，不触发重新研判。
+旧source准备状态恢复时核验原500股，不改确认标识。已知修订仍关闭采纳，资格失效仍撤下。
+普通OPEN确认、审计和指令原子提交。重复请求读回原指令，不重复下发。
+指令标识追加提议身份摘要，避免同日不同提议撞用固定序号。OPEN没有一次性计数器，其他动作路径保持。
+页面只在确认内容确实变化时清空原理由，不把连接失败叫建议更新。
+新增批次43，实际API、来源解析、筛选、快速检查和SQL事务共13例；桥逐票7例。
+47批PMS禁网回归与三页面守卫通过。桥完整版本契约回归通过。部署及真机读数随后追加。

app/db/session.py

@@ -365,3 +365,38 @@ def publish_prepared_proposal(old, candidate, verdict):
                'verdict':'NOTE','price':float(hard.get('price') or 0),'hard':encode(audit),'failed':'[]',
                'reason':'后台完成建议更新，等待人工确认','ref':old['proposal_id']})
     return 1
+
+
+def accept_open_proposal(old, hard, instruction, reason):
+    """普通建仓的确认、动作审计、待执行指令同成同败；每条SQL仍严格单表。"""
+    import json
+    from datetime import datetime
+    if old['action'] != 'OPEN' or hard.get('kind') == 'strategy' or int(old['qty']) <= 0:
+        raise ValueError('只接受有效的普通建仓建议')
+    encode = lambda v: json.dumps(v, ensure_ascii=False, default=str)
+    now = datetime.now()
+    _check_cooldown('proxy')
+    with get_engine('proxy').begin() as c:
+        def query(sql, params):
+            sql = map_tables(sql)
+            assert_single_table(sql)
+            return c.execute(text(sql), params)
+        n = query("UPDATE pms_proposal SET status='ACCEPTED',decided_at=:at,hard_numbers_json=:hn "
+                  "WHERE proposal_id=:pid AND status='WAIT_USER' AND hard_numbers_json=:old "
+                  "AND expire_at>=:at AND qty=:qty",
+                  {'at':now,'hn':encode(hard),'pid':old['proposal_id'],
+                   'old':old['hard_numbers_json'],'qty':int(old['qty'])}).rowcount
+        if not n:
+            return False
+        query('INSERT INTO pms_action_ledger (ts_code,decided_at,action,arbiter,verdict,price_at,'
+              'hard_numbers_json,failed_checks_json,reason,ref_id,outcome_scored) '
+              "VALUES (:code,:at,'OPEN','user','PASS',:px,:hn,'[]',:reason,:pid,0)",
+              {'code':old['ts_code'],'at':now,'px':float(hard.get('price') or 0),
+               'hn':encode(hard),'reason':reason[:500],'pid':old['proposal_id']})
+        query('INSERT INTO pms_instruction (instruction_id,origin_type,origin_id,ts_code,action,side,'
+              'qty,limit_price,window_tdays,status,exec_qty,progress_json,created_at,updated_at) '
+              "VALUES (:iid,'proposal',:pid,:code,'OPEN','buy',:qty,:price,:window,'PROPOSED',0,:pg,:at,:at)",
+              {'iid':instruction['id'],'pid':old['proposal_id'],'code':old['ts_code'],
+               'qty':int(old['qty']),'price':hard.get('price'),'window':instruction['window'],
+               'pg':encode(instruction['progress']),'at':now})
+    return True

app/repo/pms_repo.py

@@ -1096,3 +1096,15 @@ def list_op_log(*, limit: int = 200) -> list:
 def publish_prepared_proposal(old, candidate, verdict):
     from app.db.session import publish_prepared_proposal as publish
     return publish(old, candidate, verdict)
+
+
+def accept_open_proposal(old, hard, instruction, reason):
+    from app.db.session import accept_open_proposal as accept
+    return accept(old, hard, instruction, reason)
+
+
+def accepted_open_instruction(proposal_id):
+    rows = fetch_all("SELECT instruction_id FROM pms_instruction "
+                     "WHERE origin_type='proposal' AND origin_id=:pid AND action='OPEN' ORDER BY id",
+                     {'pid': proposal_id})
+    return rows[0]['instruction_id'] if len(rows) == 1 else None

app/services/plan_feed.py

@@ -347,7 +347,7 @@ def _rows(raw, bucket: str) -> list:
     return out
 
 
-def parse_plan(payload, *, requested=None) -> dict:
+def parse_plan(payload, *, requested=None, allow_empty=False) -> dict:
     """应答 → 内部结构。缺 date 或两档全空都算变形 (抛 PlanFeedError)。
 
     requested: 本次实际发出的 {top, obs_top} (缺省按上游签名的默认值 20/10 算)。
@@ -360,7 +360,7 @@ def parse_plan(payload, *, requested=None) -> dict:
         raise PlanFeedError("应答缺 date —— 判不了新鲜度, 按取不到处理")
     main = _rows(payload.get("main"), BUCKET_MAIN)
     observe = _rows(payload.get("observe"), BUCKET_OBSERVE)
-    if not main and not observe:
+    if not main and not observe and not allow_empty:
         raise PlanFeedError(f"计划 {date} 主榜与观察档都是空的")
     state = dict(payload.get('refresh_status') or {})
     state['pending_codes'] = [normalize_code(c) for c in state.get('pending_codes') or []]
@@ -901,11 +901,32 @@ def get_plan(*, force: bool = False, date=None) -> dict:
     return plan
 
 
-def read_current_plan():
+def read_current_plan(*, code=None):
     """采纳只读最新完整发布版，限时取数，不写接收快照或行业映射。"""
     p = _params()
+    if code is None:
         plan = fetch(base=p['base'], path=p['path'], timeout=min(3, max(1, p['timeout'])),
                      extra_params=build_query(p))
+    else:
+        # 不能回退整榜下载：那会把采纳重新变成慢请求。旧桥须先升级兼容接口。
+        import requests
+        try:
+            r = requests.get(p['base'].rstrip('/') + '/' + p['path'].strip('/') + '/check',
+                             params={'code': code}, timeout=(1, min(3, max(1, p['timeout']))))
+            r.raise_for_status()
+            payload = r.json()
+            if (not isinstance(payload, dict) or normalize_code(payload.get('check_code')) != code
+                    or not isinstance(payload.get('found'), bool)
+                    or not (payload.get('decision') or {}).get('id')
+                    or not all(isinstance(payload.get(b), list) for b in ('main', 'observe'))):
+                raise ValueError('逐票核验应答不完整')
+            rows = payload['main'] + payload['observe']
+            if (len(rows) != int(payload['found']) or
+                    any(normalize_code(row.get('code')) != code for row in rows)):
+                raise ValueError('逐票核验身份不一致')
+            plan = parse_plan(payload, allow_empty=True)
+        except Exception as e:
+            raise PlanFeedError('暂时无法核对最新资料，本次未采纳，请稍后重试') from e
     assert_fresh(plan, max_stale_tdays=p['stale_tdays'])
     return plan
 

app/services/plan_recheck.py

@@ -3,6 +3,7 @@ from datetime import datetime, timedelta
 import logging
 import re
 from app.core.plan_revision import compare, fingerprint, portfolio_context, confirmation_revision
+from app.services.plan_feed import PlanFeedError
 
 logger = logging.getLogger('pms.proposal.prepare')
 
@@ -12,7 +13,7 @@ def current_basis(hard, code, proposal_id=None, *, record=True):
     view = portfolio.positions_view()
     stock = command_service.effective_stock_params()
     params = ps._scan_params(view)
-    options = {'force': True} if record else {'record': False, 'plan_override': plan_feed.read_current_plan()}
+    options = {'force': True} if record else {'record': False, 'plan_override': plan_feed.read_current_plan(code=code)}
     selected = plan_feed.candidates(held=[x['ts_code'] for x in view['held']],
         black=[c for c, v in stock.items() if v.get('black')],
         route_by_verdict=bool(params.get('open_route_by_verdict')), recheck_code=code, **options)
@@ -161,6 +162,10 @@ def before_accept(p, expected_revision):
         checked = check['provenance']
         version_check = dict(checked.get('version_check') or {}, checked_decision=checked.get('decision'))
         return {'ok': True, 'proposal': dict(p, hard_numbers=dict(hard, version_check=version_check))}
+    except PlanFeedError as e:
+        # 传输故障没有改变研究或确认内容。保留已准备状态，不制造后台重算。
+        logger.warning('采纳来源核验失败 proposal=%s: %s', p.get('proposal_id'), e, exc_info=True)
+        return {'ok': False, 'retryable': True, 'error': str(e), 'proposal': p}
     except Exception:
         logger.exception('采纳快速检查失败 proposal=%s', p.get('proposal_id'))
         reason = '当前资料暂不可用，建议保留，后台将自动检查'
@@ -182,6 +187,9 @@ def refresh_pending(*, now=None, deadline=None):
             continue
         try:
             result = refresh_one(p, now=now, deadline=deadline)
+        except PlanFeedError as e:
+            logger.warning('后台核对暂不可用 proposal=%s: %s', p['proposal_id'], e)
+            result = {'ok': False, 'error': str(e), 'retryable': True}
         except Exception:
             logger.exception('后台更新建议失败 proposal=%s', p['proposal_id'])
             result = _pending_response(p, '建议更新暂未完成，系统将自动重试', 'source')
@@ -206,6 +214,17 @@ def refresh_one(p, *, now=None, deadline=None):
         return _pending_response(p, check.get('error') or '建议资料暂不可用', check.get('stage', 'source'))
     if check.get('ok') and (hard.get('preparation') or {}).get('status') == 'ready' and not hard.get('last_accept_check'):
         return {'ok': True}
+    # 旧的短暂取数失败不应触发扫描、重新选数和新消费；当前原数量仍过闸即可恢复。
+    if ((hard.get('preparation') or {}).get('stage') in ('source', 'research_pending')
+            and hard.get('consumption_id') and hard.get('proposal_revision')
+            and hard.get('prepared_judgement') and hard.get('plan_basis') is not None
+            and not hard.get('last_accept_check') and not check.get('research_changed')
+            and not check.get('params_changed') and not fast_checks(p, check)):
+        restored = ready_hard(hard, _judgement(p), now)
+        candidate = {'qty': p['qty'], 'hard_numbers': restored}
+        if pms_repo.publish_prepared_proposal(p, candidate, _judgement(p)):
+            return {'ok': True, 'proposal': dict(p, hard_numbers=restored)}
+        return {'ok': False, 'error': '建议已被其他操作更新'}
     research_changed = check.get('research_changed', check.get('material', False))
     previous_failure = hard.get('last_accept_check') or {}
     needs_model = research_changed or previous_failure.get('stage') in ('judge', 'system')

app/web/main.py

@@ -519,6 +519,12 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default=
         p = pms_repo.get_proposal(proposal_id)
         if not p:
             return {"ok": False, "error": "提议不存在"}
+        if (decision == 'ACCEPTED' and p['status'] == 'ACCEPTED' and p['action'] == 'OPEN'
+                and (p.get('hard_numbers') or {}).get('kind') != 'strategy'):
+            existing = pms_repo.accepted_open_instruction(proposal_id)
+            if existing:
+                return {'ok': True, 'decision': decision, 'instruction_id': existing, 'already_accepted': True}
+            return {'ok': False, 'error': '该建议已有采纳记录，但指令记录不完整，请联系维护人员核对'}
         if p["status"] != "WAIT_USER":
             check = (p.get('hard_numbers') or {}).get('last_accept_check') or {}
             if p['status'] == 'EXPIRED' and check.get('status') == 'withdrawn':
@@ -550,6 +556,27 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default=
                 return checked
             expected_hard = p.get('hard_numbers_json')
             p = checked['proposal']
+        if (decision == 'ACCEPTED' and p['action'] == 'OPEN'
+                and (p.get('hard_numbers') or {}).get('kind') != 'strategy'):
+            # 先完整准备写入内容，再一次提交。不能先改成已采纳，再独立写账与落指令。
+            import hashlib
+            hn = dict(p.get('hard_numbers') or {}, decide_reason_code=reason_code)
+            window = param_store.get_int('PMS_EXEC_WINDOW_TDAYS', 3)
+            iid = cs.make_instruction_id(td.ymd(), p['ts_code'], 'OPEN', 1)
+            iid += '_' + hashlib.sha256(proposal_id.encode()).hexdigest()[:12]
+            instruction = {'id': iid, 'window': window, 'progress': {
+                'from_proposal': proposal_id, 'is_command': False, 'children': [],
+                'plan': {k: hn.get(k) for k in ('decision', 'consumption_id', 'review_ref',
+                                               'execution_rank', 'priority_reason', 'version_check')},
+                'deadline': str(td.window_deadline(None, window))}}
+            if pms_repo.accept_open_proposal(p, hn, instruction, reason):
+                return {'ok': True, 'decision': decision, 'instruction_id': iid}
+            latest = pms_repo.get_proposal(proposal_id)
+            existing = (pms_repo.accepted_open_instruction(proposal_id)
+                        if latest and latest['status'] == 'ACCEPTED' else None)
+            if existing:
+                return {'ok': True, 'decision': decision, 'instruction_id': existing, 'already_accepted': True}
+            return {'ok': False, 'error': '建议已更新或过期，请查看当前内容', 'proposal': latest}
         changed = (pms_repo.decide_proposal(proposal_id, decision, expected_hard=expected_hard)
                    if expected_hard is not None else pms_repo.decide_proposal(proposal_id, decision))
         if not changed:

app/web/static/index.html

@@ -4431,8 +4431,10 @@ createApp({
             ? '尚未收到处理结果。请关闭后刷新，先核对待确认提议和当日指令，勿立即重复提交。'
             : (d.error || '本次未能采纳，请稍后再试');
           if (d.proposal) {
+            const priorRevision = (dd.row.hard_numbers || {}).proposal_revision;
             dd.row = { ...dd.row, ...d.proposal };
-            dd.updated = !dd.withdrawn && !dd.pending && !((d.proposal.hard_numbers||{}).last_accept_check);
+            dd.updated = !d.retryable && !dd.withdrawn && !dd.pending &&
+              priorRevision !== (d.proposal.hard_numbers || {}).proposal_revision;
             if (dd.updated) { dd.code = ''; dd.note = ''; }
             await loadProposals();
           }

scripts/run_tests.py

@@ -5,6 +5,7 @@
 运行:  docker compose run --rm pms-web python scripts/run_tests.py
 
 包含 (例数按 2026-10-12 隔离回归实跑校正):
+  test_batch43_units.py 逐票来源核验、实际API至SQL事务、故障回滚与幂等 (13 例)
   test_batch42_units.py 提议先准备后采纳、模型复用、并发更新与发布审计 (24 例)
   test_batch41_units.py 跨流程买卖互斥、清仓收尾、出口撤单与真实进展及人工清仓T+1 (24 例)
   test_batch40_units.py 逐笔历史修复、进度暂停、并发与事务回滚与均价失效 (22 例)
@@ -193,7 +194,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
           "test_batch35_units.py",
           "test_batch36_units.py",
           "test_batch37_units.py",
-          "test_batch38_units.py", "test_batch39_units.py", "test_batch40_units.py", "test_batch41_units.py", "test_batch42_units.py",
+          "test_batch38_units.py", "test_batch39_units.py", "test_batch40_units.py", "test_batch41_units.py", "test_batch42_units.py", "test_batch43_units.py",
           "test_page_copy_guard.py",
           "test_page_enum_guard.py",
           "test_page_wiring_guard.py",

scripts/test_batch43_units.py

@@ -0,0 +1,145 @@
+"""开发机：python scripts/unit_test_guard.py scripts/test_batch43_units.py。隔离禁网，真实SQL事务。"""
+import unit_test_guard
+import os,sys,copy,tempfile,json
+from datetime import datetime,timedelta
+from types import SimpleNamespace
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import patch
+import unittest
+sys.path.insert(0,os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from sqlalchemy import create_engine,text,event
+from app.db import session
+from app.repo import pms_repo
+from app.services import plan_feed as pf,plan_recheck as rc,proposal_service as ps,portfolio,command_service,param_store,market,execution_conflicts
+from app.core import plan_revision as pr,tradedays as td
+from app.web import main
+import requests
+
+CODE='300693.SZ'
+
+class AcceptancePathTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
+        self.engine=create_engine('sqlite:///'+self.tmp.name+'/state.sqlite',connect_args={'check_same_thread':False})
+        self.addCleanup(self.engine.dispose)
+        def use(target,**kw):
+            p=patch(target,**kw);v=p.start();self.addCleanup(p.stop);return v
+        use('app.db.session.get_engine',return_value=self.engine);use('app.db.session._check_cooldown')
+        use('app.services.param_store.get',side_effect=lambda k,d=None:d)
+        use('app.services.param_store.get_int',side_effect=lambda k,d=0:d)
+        use('app.services.param_store.get_float',side_effect=lambda k,d=0:d)
+        use('app.services.param_store.get_list',side_effect=lambda k,d=None:d or [])
+        use('app.services.param_store.get_bool',side_effect=lambda k,d=False:True if k=='PMS_PLAN_VERSION_RECHECK' else d)
+        use('app.web.main._oplog');use('app.web.main._actor',return_value='test')
+        self.params=pf._params();self.params.update(base='http://bridge',tiers=['强传导'])
+        use('app.services.plan_feed._params',return_value=self.params)
+        self.view={'held':[],'params':{'max_names':40,'scale':2000000,'portfolio_cap':0.7},
+                   'totals':{'names_count':0,'portfolio_mv':0,'cash_source':portfolio.CASH_WS,'cash_avail':100000}}
+        self.commitments={'names':set(),'amount':0,'basis':[]}
+        self.scan={'open_route_by_verdict':True}
+        use('app.services.portfolio.positions_view',return_value=self.view)
+        use('app.services.command_service.effective_stock_params',return_value={})
+        use('app.services.proposal_service._scan_params',return_value=self.scan)
+        use('app.services.proposal_service._inflight_open_commitments',return_value=self.commitments)
+        use('app.services.execution_conflicts.buy_blocker',return_value=None)
+        use('app.services.market.get_price',return_value=34.18)
+        use('app.services.market.day_snapshot',return_value={});use('app.services.market.get_ma5',return_value=34)
+        use('app.services.proposal_service.check_candidate_rules',return_value={'passed':True})
+        use('app.core.tradedays.window_deadline',return_value='2026-10-23')
+        use('app.services.proposal_service._scan_open',side_effect=AssertionError('不能重扫'))
+        use('app.services.proposal_service._route_one',side_effect=AssertionError('不能重问模型'))
+        use('app.repo.pms_repo.insert_plan_snapshot',side_effect=AssertionError('采纳不能写新消费'))
+        self.raw={'date':datetime.now().date().isoformat(),'schema_version':2,'decision':{'id':'live'},
+            'check_code':'SZ300693','found':True,'refresh_status':{'pending_codes':[]},
+            'main':[{'code':'SZ300693','name':'盛弘股份','score':230,'tier':'强传导','rank':555,
+                     'verdict':'候选','risk':[],'company_review_text':'研究依据'}],'observe':[]}
+        def response(url,**kw):
+            self.assertEqual(url,'http://bridge/plan/check');self.assertEqual(kw['params'],{'code':CODE})
+            return SimpleNamespace(raise_for_status=lambda:None,json=lambda:copy.deepcopy(self.raw))
+        self.http=use('requests.get',side_effect=response)
+        selected=pf.candidates(record=False,recheck_code=CODE,plan_override=pf.parse_plan(self.raw),route_by_verdict=True)
+        h={'price':34.18,'sector':'电气设备','preparation':{'status':'ready'},'proposal_revision':'shown',
+           'consumption_id':'already-saved','decision':{'id':'prepared'},'plan_basis':pr.basis(selected['items'][0]),
+           'scan_params_hash':pr.fingerprint(self.scan),'portfolio_context_hash':pr.portfolio_context(self.view,self.commitments),
+           'prepared_judgement':{'verdict':'UNAVAILABLE','reason':'原人工确认依据','degraded':True}}
+        self.hard=h;self.pid='PRP_20261021_300693SZ_OPEN'
+        ddl=[
+          'CREATE TABLE pms_proposal(id INTEGER PRIMARY KEY,proposal_id TEXT UNIQUE,ts_code TEXT,action TEXT,qty INTEGER,hard_numbers_json TEXT,status TEXT,expire_at TEXT,decided_at TEXT,judge_verdict TEXT,judge_reason TEXT,archived_at TEXT)',
+          'CREATE TABLE pms_action_ledger(ts_code TEXT,decided_at TEXT,action TEXT,arbiter TEXT,verdict TEXT,price_at REAL,hard_numbers_json TEXT,failed_checks_json TEXT,reason TEXT,ref_id TEXT,outcome_scored INTEGER)',
+          'CREATE TABLE pms_instruction(id INTEGER PRIMARY KEY,instruction_id TEXT UNIQUE,origin_type TEXT,origin_id TEXT,ts_code TEXT,action TEXT,side TEXT,qty INTEGER,limit_price REAL,window_tdays INTEGER,status TEXT,exec_qty INTEGER,progress_json TEXT,created_at TEXT,updated_at TEXT)']
+        with self.engine.begin() as c:
+            for sql in ddl:c.execute(text(sql))
+            c.execute(text("INSERT INTO pms_proposal(proposal_id,ts_code,action,qty,hard_numbers_json,status,expire_at) VALUES (:pid,:code,'OPEN',500,:hn,'WAIT_USER',:expiry)"),
+                {'pid':self.pid,'code':CODE,'hn':pms_repo._dumps(h),'expiry':datetime.now()+timedelta(hours=1)})
+    def decide(self,revision='shown'):
+        return main.api_decide(None,self.pid,{'decision':'ACCEPTED','reason':'用户确认','proposal_revision':revision})
+    def counts(self):
+        with self.engine.connect() as c:return tuple(c.execute(text('SELECT COUNT(*) FROM '+t)).scalar() for t in ('pms_instruction','pms_action_ledger'))
+    def test_full_api_to_sql_accepts_prepared_quantity_and_pairs_original_source(self):
+        r=self.decide();self.assertTrue(r['ok'],r);self.assertEqual(self.counts(),(1,1))
+        with self.engine.connect() as c:
+            row=dict(c.execute(text('SELECT * FROM pms_instruction')).mappings().one())
+        self.assertEqual(row['qty'],500);p=json.loads(row['progress_json'])['plan']
+        self.assertEqual(p['consumption_id'],'already-saved');self.assertEqual(p['decision']['id'],'prepared')
+        self.assertEqual(p['version_check']['checked_decision']['id'],'live')
+    def test_transport_timeout_keeps_ready_and_retry_uses_same_confirmation(self):
+        self.http.side_effect=requests.exceptions.ReadTimeout('3 seconds')
+        r=self.decide();self.assertFalse(r['ok']);self.assertTrue(r['retryable']);self.assertEqual(self.counts(),(0,0))
+        self.assertEqual(pms_repo.get_proposal(self.pid)['hard_numbers']['preparation']['status'],'ready')
+        self.http.side_effect=lambda *a,**k:SimpleNamespace(raise_for_status=lambda:None,json=lambda:self.raw)
+        self.assertTrue(self.decide()['ok']);self.assertEqual(self.counts(),(1,1))
+    def test_response_lost_retry_returns_exact_existing_instruction(self):
+        a=self.decide();self.http.side_effect=AssertionError('重试已采纳不再查来源')
+        b=self.decide();self.assertTrue(b['already_accepted']);self.assertEqual(a['instruction_id'],b['instruction_id'])
+        self.assertEqual(self.counts(),(1,1))
+    def test_two_concurrent_confirmations_create_one_instruction_and_audit(self):
+        with ThreadPoolExecutor(2) as pool:r=list(pool.map(lambda _:self.decide(),range(2)))
+        self.assertTrue(all(x['ok'] for x in r),r);self.assertEqual(len({x['instruction_id'] for x in r}),1)
+        self.assertEqual(self.counts(),(1,1))
+    def test_instruction_or_audit_failure_rolls_back_confirmation(self):
+        for table in ('pms_instruction','pms_action_ledger'):
+            def fail(conn,cursor,statement,params,context,many):
+                if statement.startswith('INSERT INTO '+table):raise RuntimeError('故障注入')
+            event.listen(self.engine,'before_cursor_execute',fail)
+            try:r=self.decide()
+            finally:event.remove(self.engine,'before_cursor_execute',fail)
+            self.assertFalse(r['ok']);self.assertEqual(self.counts(),(0,0))
+            self.assertEqual(pms_repo.get_proposal(self.pid)['status'],'WAIT_USER')
+        self.assertTrue(self.decide()['ok'])
+    def test_stale_confirmation_cannot_accept(self):
+        self.assertFalse(self.decide('old')['ok']);self.assertEqual(self.counts(),(0,0))
+    def test_unknown_shape_or_wrong_stock_does_not_withdraw_or_accept(self):
+        for bad in ({},dict(self.raw,check_code='SH600000'),dict(self.raw,found=False),dict(self.raw,main=None)):
+            self.http.side_effect=lambda *a,**k:SimpleNamespace(raise_for_status=lambda:None,json=lambda:bad)
+            self.assertFalse(self.decide()['ok']);self.assertEqual(self.counts(),(0,0))
+            self.assertEqual(pms_repo.get_proposal(self.pid)['status'],'WAIT_USER')
+    def test_known_research_pending_is_not_transport_retry(self):
+        self.raw['refresh_status']['pending_codes']=['SZ300693']
+        r=self.decide();self.assertTrue(r['pending']);self.assertEqual(self.counts(),(0,0))
+        self.assertEqual(pms_repo.get_proposal(self.pid)['hard_numbers']['preparation']['status'],'pending')
+    def test_missing_qualification_withdraws(self):
+        self.raw.update(found=False,main=[])
+        r=self.decide();self.assertTrue(r['withdrawn'],r);self.assertEqual(self.counts(),(0,1))
+        self.assertEqual(pms_repo.get_proposal(self.pid)['status'],'EXPIRED')
+    def test_material_change_waits_for_background_not_accept(self):
+        self.raw['main'][0]['risk']=['重大风险变化']
+        self.assertTrue(self.decide()['pending']);self.assertEqual(self.counts(),(0,0))
+    def test_existing_quantity_still_requires_funds(self):
+        self.view['totals']['cash_avail']=100
+        r=self.decide();self.assertFalse(r['ok']);self.assertIn('资金不足',r['error']);self.assertEqual(self.counts(),(0,0))
+    def test_source_recovery_restores_original_content_without_scan(self):
+        p=pms_repo.get_proposal(self.pid);h=dict(p['hard_numbers'],preparation={'status':'pending','stage':'source'})
+        pms_repo.record_proposal_check(p,h)
+        self.assertTrue(rc.refresh_one(pms_repo.get_proposal(self.pid))['ok'])
+        after=pms_repo.get_proposal(self.pid)
+        self.assertEqual(after['qty'],500);self.assertEqual(after['hard_numbers']['proposal_revision'],'shown')
+        self.assertTrue(self.decide()['ok']);self.assertEqual(self.counts(),(1,2))
+    def test_expiry_during_checks_prevents_commit(self):
+        original=rc.fast_checks
+        def expire(p,c):
+            session.execute("UPDATE pms_proposal SET expire_at=:at WHERE proposal_id=:pid",{'at':datetime.now()-timedelta(seconds=1),'pid':self.pid})
+            return original(p,c)
+        with patch.object(rc,'fast_checks',side_effect=expire):r=self.decide()
+        self.assertFalse(r['ok']);self.assertEqual(self.counts(),(0,0))
+
+if __name__=='__main__':unittest.main()