feat: keep proxy forwarding responsive under capture pressure

README.md

@@ -47,7 +47,7 @@ Rust crate 位于 `crates/prompt-ir`，公开 JSON Schema 位于 `schemas/prompt
 cargo run -p codeischeap-desktop-api --bin export-desktop-contract
 ```
 
-捕获 sidecar 位于 `sidecars/mitmproxy`，跨进程契约位于 `crates/capture-ipc`，公开 CaptureEnvelope Schema 位于 `schemas/capture-envelope/v0.1.schema.json`。sidecar 的安装、测试与打包命令见 [`sidecars/mitmproxy/README.md`](./sidecars/mitmproxy/README.md)。
+捕获 sidecar 位于 `sidecars/mitmproxy`，跨进程契约位于 `crates/capture-ipc`，公开 CaptureEnvelope Schema 位于 `schemas/capture-envelope/v0.1.schema.json`。打包探针会验证 HTTP/1.1、HTTP/2、压缩、流式脱敏、客户端取消与捕获 IPC 背压；捕获队列满时只丢弃记录，不阻塞代理转发。sidecar 的安装、测试与打包命令见 [`sidecars/mitmproxy/README.md`](./sidecars/mitmproxy/README.md)。
 
 捕获范围与敏感字段由 `policies/capture-policy.v0.1.json` 定义，公开 schema 位于 `schemas/capture-policy/v0.1.schema.json`。Python sidecar 在 IPC 前执行策略，`crates/core` 在进入持久化前再次拒绝越界请求并删除遗漏凭据。
 

crates/sidecar-runtime/src/lib.rs

@@ -1592,6 +1592,8 @@ struct IntegrationProbe {
     non_target_tunnel: bool,
     http2_preserved: bool,
     transport_context_preserved: bool,
+    client_cancellation_survived: bool,
+    capture_backpressure_nonblocking: bool,
 }
 
 fn validate_manifest(
@@ -1650,6 +1652,8 @@ fn validate_manifest(
         || !probe.non_target_tunnel
         || !probe.http2_preserved
         || !probe.transport_context_preserved
+        || !probe.client_cancellation_survived
+        || !probe.capture_backpressure_nonblocking
         || probe.credential_canaries_in_envelope != 0
     {
         return Err(SidecarError::InvalidManifest(
@@ -2196,7 +2200,9 @@ MAoGCCqGSM49BAMCA0gAMEUCIQC1PB8+NumezrQf5unFGhVeufUcyw/sjH6p1aqs
                 "stream_credentials_removed": true,
                 "non_target_tunnel": true,
                 "http2_preserved": true,
-                "transport_context_preserved": true
+                "transport_context_preserved": true,
+                "client_cancellation_survived": true,
+                "capture_backpressure_nonblocking": true
             },
             "bundle_ready": true,
             "release_ready": signature == "valid"

docs/progress.html

@@ -62,7 +62,7 @@ <h2 id="streams">4. 工作流进度</h2>
         <tr><td>Capture / Network</td><td>CAP-001~007</td><td>85%</td><td><span class="status active">In progress</span></td><td>Codex / TBD</td><td>补齐 macOS 特权代理 helper；生产签名随发布凭据补齐</td></tr>
         <tr><td>Prompt / Adapters</td><td>PAR-001~007</td><td>100%</td><td><span class="status done">Done</span></td><td>Codex / TBD</td><td>保持价格目录与 provider usage 映射可追溯</td></tr>
         <tr><td>Data / Security</td><td>DAT-001~002、SEC-001~004</td><td>75%</td><td><span class="status active">In progress</span></td><td>Codex / TBD</td><td>继续 OS 级 IPC ACL、WASI 权限、供应链控制与独立安全评审</td></tr>
-        <tr><td>Test / Release</td><td>TST-001~004、REL-001~003</td><td>48%</td><td><span class="status active">In progress</span></td><td>Codex / TBD</td><td>继续 TST-002 Proxy 取消/背压与支持处理流程</td></tr>
+        <tr><td>Test / Release</td><td>TST-001~004、REL-001~003</td><td>55%</td><td><span class="status active">In progress</span></td><td>Codex / TBD</td><td>继续 TST-002 完整协议一致性与支持处理流程</td></tr>
       </tbody></table></div>
 
       <h2 id="current-sprint">5. 当前迭代：S5 / S6</h2>
@@ -98,7 +98,7 @@ <h2 id="current-sprint">5. 当前迭代：S5 / S6</h2>
         <tr><td class="task-id">PAR-006</td><td>Ollama 适配器</td><td><span class="status done">Done</span></td><td>100%</td><td>Codex / TBD</td><td>2026-07-17</td><td>本地 /api/chat 与 /api/generate、system/messages/images/tools/options、JSON/NDJSON 响应、usage 与 Raw 精确定位完成</td></tr>
         <tr><td class="task-id">PAR-007</td><td>token、成本和语义指纹</td><td><span class="status done">Done</span></td><td>100%</td><td>Codex / TBD</td><td>2026-07-17</td><td>四厂商 reported usage 归一化、显式 estimated 估算、版本化价格匹配、未知价格留空与 BLAKE3-256 语义指纹完成</td></tr>
         <tr><td class="task-id">TST-001</td><td>协议 fixture 与 golden tests</td><td><span class="status done">Done</span></td><td>100%</td><td>Codex / TBD</td><td>2026-07-17</td><td>版本化能力矩阵覆盖 OpenAI、Anthropic、Gemini 与 Ollama 的请求、响应、流式、工具、多模态、错误和 Raw fallback；声明均由 fixture 与 golden 验证</td></tr>
-        <tr><td class="task-id">TST-002</td><td>Gateway/Proxy 集成测试</td><td><span class="status active">In progress</span></td><td>65%</td><td>Codex / TBD</td><td>2026-08-25</td><td>认证 IPC、暂停丢弃、真实模式切换与进程树清理、压缩、流式脱敏、非目标 TLS、HTTP/2 基线一致性及 Windows CA 精确增删通过；真实 ROOT 往返需交互式 Windows 会话，Proxy 取消、背压与完整协议一致性待扩展</td></tr>
+        <tr><td class="task-id">TST-002</td><td>Gateway/Proxy 集成测试</td><td><span class="status active">In progress</span></td><td>82%</td><td>Codex / TBD</td><td>2026-08-25</td><td>认证 IPC、暂停丢弃、真实模式切换与进程树清理、压缩、流式脱敏、非目标 TLS、HTTP/2 基线一致性及 Windows CA 精确增删通过；真实 sidecar 在客户端取消后保持存活，IPC 停止消费且 72 个事件超过容量 64 时，36 个目标响应仍在 15 秒内完成；真实 ROOT 往返和其余协议一致性待扩展</td></tr>
         <tr><td class="task-id">REL-002</td><td>诊断与支持包</td><td><span class="status active">In progress</span></td><td>75%</td><td>Codex / Support Owner TBD</td><td>2026-09-22</td><td>版本化 JSON 支持包可预览、复制和保存，包含不带请求标识的兼容诊断树；256 KiB code-only journal 与最近 100 条事件接入，排除 Prompt、Raw 和日志详情；支持处理流程待完成</td></tr>
         <tr><td class="task-id">SPIKE-001</td><td>Gateway 流式透明转发验证</td><td><span class="status done">Done</span></td><td>100%</td><td>Codex / TBD</td><td>2026-07-14</td><td>双向流式、取消传递、头清理与稳定 502 集成测试通过</td></tr>
         <tr><td class="task-id">SPIKE-002</td><td>mitmproxy sidecar IPC/打包验证</td><td><span class="status done">Done</span></td><td>100%</td><td>Codex / TBD</td><td>2026-07-14</td><td><a href="spikes/mitmproxy-sidecar.html">凭据清理、IPC、打包与真实转发通过</a></td></tr>
@@ -121,7 +121,7 @@ <h2 id="backlog">6. 后续迭代承诺</h2>
         <tr><td>S1</td><td>APP-001/002、DAT-001、SEC-001、Core 事件骨架</td><td><span class="status active">In progress</span></td><td>加密桌面链路与 SEC-001 已通过；Core event 与独立安全评审待完成</td></tr>
         <tr><td>S2</td><td>CAP-001/002、PAR-002/003、APP-003</td><td><span class="status done">Done</span></td><td>Gateway、OpenAI 解析与千条实时工作台全部通过验收</td></tr>
         <tr><td>S3</td><td>PAR-004、APP-004、TST-001、DAT-002</td><td><span class="status done">Done</span></td><td>双厂商 Inspector、能力矩阵与数据生命周期全部通过验收</td></tr>
-        <tr><td>S4</td><td>CAP-003~005、TST-002</td><td><span class="status active">In progress</span></td><td>sidecar bundle、桌面运行时、协议矩阵、跨平台 CA 状态及两平台用户级信任生命周期已实现；签名、Proxy 取消/背压与交互式验收待推进</td></tr>
+        <tr><td>S4</td><td>CAP-003~005、TST-002</td><td><span class="status active">In progress</span></td><td>sidecar bundle、桌面运行时、协议矩阵、Proxy 取消/背压、跨平台 CA 状态及两平台用户级信任生命周期已实现；签名、其余协议一致性与交互式验收待推进</td></tr>
         <tr><td>S5</td><td>CAP-006/007、SEC-002/003、APP-006、TST-003</td><td><span class="status active">In progress</span></td><td>SEC-002、CAP-007 完成，IPC 抗阻塞、双模式 OS PID 归因和兼容诊断树已接入；继续 macOS helper、OS socket ACL 与其余故障注入</td></tr>
         <tr><td>S6</td><td>PAR-005~007、APP-005</td><td><span class="status done">Done</span></td><td>四厂商适配器、token/成本/指纹、全文搜索和结构/文本 Compare 全部完成</td></tr>
         <tr><td>S7</td><td>TST-004、性能、可访问性、诊断与保留</td><td><span class="status">Not started</span></td><td>功能冻结</td></tr>
@@ -218,6 +218,7 @@ <h2 id="decisions">11. 决策与变更记录</h2>
         <tr><td>2026-07-18</td><td>Capture / Security</td><td>CAP-007 完成、SEC-003 推进</td><td>IPC 0.3 在认证帧中传递严格 loopback 临时端点，桌面端忽略 sidecar PID 并用 OS socket 表精确归因 Proxy 进程；端点不进入 Envelope、数据库或导出，查询失败保持未知</td><td>Codex</td></tr>
         <tr><td>2026-07-18</td><td>Security / IPC</td><td>SEC-003 sidecar peer 绑定完成首段</td><td>IPC 0.4 让 sidecar 等待服务端 ACK，从协议层保持首连接直到 Windows/macOS OS socket 表完成 PID 核对；不匹配、查询失败或 2 秒超时只拒绝捕获，代理转发继续，后续连接仍受会话 token 保护</td><td>Codex</td></tr>
         <tr><td>2026-07-18</td><td>Security / IPC</td><td>SEC-003 Linux peer 绑定完成</td><td>Linux 读取当前 network namespace 的 /proc/net/tcp{,6}，以客户端到 IPC 服务端的精确四元组定位 ESTABLISHED socket inode，再扫描 /proc/&lt;pid&gt;/fd；仅唯一 owner 且 PID 等于已启动 sidecar 时接受首帧，权限受限、连接消失、歧义和解析失败均保持拒绝</td><td>Codex</td></tr>
+        <tr><td>2026-07-18</td><td>Test / Proxy</td><td>TST-002 取消与背压矩阵完成</td><td>单元测试以阻塞 IPC worker 和容量 1 队列证明满队列 submit 立即失败；首次 IPC 失败会清空 backlog、熔断 10 秒并自动探测恢复；真实打包 sidecar 在 IPC listener 停止消费后承受 36 个目标请求、72 个捕获事件和一次客户端取消，转发响应保持一致且进程继续存活；两个结论成为 Python 与 Rust bundle 强制合同</td><td>Codex</td></tr>
         <tr><td colspan="5">后续范围、架构、日期或资源变化均在此追加，并链接对应 ADR/会议结论。</td></tr>
       </tbody></table></div>
 

sidecars/mitmproxy/README.md

@@ -12,6 +12,8 @@ CIC_CAPTURE_HOSTS=api.openai.com,api.anthropic.com
 
 The addon loads `policies/capture-policy.v0.1.json` and only records exact target hosts, approved POST paths, and methods. `CIC_CAPTURE_HOSTS` can opt an OpenAI-compatible host into the same approved path set; it does not disable path checks. The addon removes credential headers, sensitive query fields, and recursively named JSON secret fields before sending an authenticated NDJSON envelope. Unsupported request body formats are omitted. The original network request is not modified.
 
+Capture delivery uses a bounded non-blocking queue. A full queue drops only the capture event. The first IPC delivery failure discards the pending backlog and opens a 10-second retry circuit; proxy forwarding continues, and the next event after the cooldown probes IPC recovery.
+
 ```powershell
 python -m pip install -r sidecars/mitmproxy/requirements-build.txt
 python -m unittest discover -s sidecars/mitmproxy/tests -v

sidecars/mitmproxy/codeischeap_addon.py

@@ -550,23 +550,49 @@ def send_envelope(
     ).encode("utf-8")
     with socket.create_connection((config.host, config.port), timeout=config.timeout_seconds) as connection:
         connection.sendall(frames)
-        acknowledgement = connection.makefile("rb").readline(256)
-        if json.loads(acknowledgement) != {"status": "accepted"}:
+        acknowledgement = bytearray()
+        while len(acknowledgement) < 256 and b"\n" not in acknowledgement:
+            chunk = connection.recv(256 - len(acknowledgement))
+            if not chunk:
+                break
+            acknowledgement.extend(chunk)
+        frame, separator, _ = bytes(acknowledgement).partition(b"\n")
+        if not separator or json.loads(frame) != {"status": "accepted"}:
             raise ValueError("capture IPC acknowledgement is invalid")
 
 
 class IpcEmitter:
-    def __init__(self, config: IpcConfig, capacity: int = 64) -> None:
+    def __init__(
+        self,
+        config: IpcConfig,
+        capacity: int = 64,
+        retry_delay_seconds: float = 10.0,
+    ) -> None:
         self._config = config
+        self._retry_delay_seconds = retry_delay_seconds
         self._queue: queue.Queue[
             tuple[dict[str, Any], dict[str, str] | None] | None
         ] = queue.Queue(maxsize=capacity)
+        self._state_lock = threading.Lock()
+        self._retry_after = 0.0
+        self._delivery_unavailable = threading.Event()
         self._worker = threading.Thread(target=self._run, name="codeischeap-ipc", daemon=True)
         self._worker.start()
 
     def submit(
         self, envelope: dict[str, Any], transport: dict[str, str] | None
     ) -> bool:
+        now = time.monotonic()
+        with self._state_lock:
+            if now < self._retry_after:
+                unavailable = True
+            else:
+                if self._retry_after:
+                    self._retry_after = 0.0
+                    self._delivery_unavailable.clear()
+                unavailable = False
+        if unavailable:
+            return False
         try:
             self._queue.put_nowait((envelope, transport))
             return True
@@ -589,15 +615,28 @@ def _run(self) -> None:
             try:
                 send_envelope(self._config, envelope, transport)
             except (OSError, ValueError):
-                _warn("CodeIsCheap capture IPC delivery failed; the request was not recorded")
+                with self._state_lock:
+                    self._retry_after = time.monotonic() + self._retry_delay_seconds
+                    self._delivery_unavailable.set()
+                if self._discard_pending():
+                    return
+
+    def _discard_pending(self) -> bool:
+        while True:
+            try:
+                submission = self._queue.get_nowait()
+            except queue.Empty:
+                return False
+            if submission is None:
+                return True
 
 
 def _warn(message: str) -> None:
     try:
         from mitmproxy import ctx
 
         ctx.log.warn(message)
-    except (ImportError, RuntimeError):
+    except (AttributeError, ImportError, RuntimeError):
         pass
 
 
@@ -638,8 +677,7 @@ def _submit(
         except (AttributeError, TypeError, ValueError):
             _warn(failure_message)
             return
-        if not self._emitter.submit(envelope, build_transport_context(flow)):
-            _warn("CodeIsCheap capture queue is full; the request was not recorded")
+        self._emitter.submit(envelope, build_transport_context(flow))
 
     def request(self, flow: Any) -> None:
         self._submit(

sidecars/mitmproxy/package_sidecar.py

@@ -299,6 +299,8 @@ def main() -> None:
             "non_target_tunnel",
             "http2_preserved",
             "transport_context_preserved",
+            "client_cancellation_survived",
+            "capture_backpressure_nonblocking",
         )
     ) and probe_result.get("credential_canaries_in_envelope") == 0
     manifest = {

sidecars/mitmproxy/tests/test_addon.py

@@ -7,6 +7,7 @@
 import sys
 import tempfile
 import threading
+import time
 import unittest
 from unittest.mock import patch
 
@@ -17,6 +18,7 @@
 sys.path.insert(0, str(SIDECAR_DIR))
 
 from codeischeap_addon import (
+    IpcEmitter,
     IpcConfig,
     build_envelope,
     build_failure_envelope,
@@ -398,7 +400,9 @@ def accept() -> None:
             with connection:
                 stream = connection.makefile("rb")
                 received.extend([stream.readline(), stream.readline()])
-                connection.sendall(b'{"status":"accepted"}\n')
+                connection.sendall(b'{"status":')
+                time.sleep(0.01)
+                connection.sendall(b'"accepted"}\n')
 
         worker = threading.Thread(target=accept)
         worker.start()
@@ -420,6 +424,90 @@ def accept() -> None:
         self.assertNotIn("transport", captured)
         self.assertEqual(captured, envelope)
 
+    def test_ipc_queue_backpressure_never_blocks_flow_submission(self) -> None:
+        first_delivery_started = threading.Event()
+        release_first_delivery = threading.Event()
+        second_delivery_finished = threading.Event()
+        deliveries: list[dict[str, object]] = []
+
+        def blocked_send(
+            config: IpcConfig,
+            envelope: dict[str, object],
+            transport: dict[str, str] | None = None,
+        ) -> None:
+            del config, transport
+            deliveries.append(envelope)
+            if len(deliveries) == 1:
+                first_delivery_started.set()
+                release_first_delivery.wait(2)
+            elif len(deliveries) == 2:
+                second_delivery_finished.set()
+
+        config = IpcConfig("127.0.0.1", 1, "synthetic-token")
+        with patch("codeischeap_addon.send_envelope", side_effect=blocked_send):
+            emitter = IpcEmitter(config, capacity=1)
+            try:
+                self.assertTrue(emitter.submit({"capture_id": "first"}, None))
+                self.assertTrue(first_delivery_started.wait(1))
+                self.assertTrue(emitter.submit({"capture_id": "queued"}, None))
+
+                started = time.monotonic()
+                self.assertFalse(emitter.submit({"capture_id": "dropped"}, None))
+                self.assertLess(time.monotonic() - started, 0.25)
+            finally:
+                release_first_delivery.set()
+
+            self.assertTrue(second_delivery_finished.wait(1))
+            emitter.close()
+            self.assertEqual(
+                [delivery["capture_id"] for delivery in deliveries],
+                ["first", "queued"],
+            )
+
+    def test_ipc_failure_discards_backlog_and_opens_a_retry_circuit(self) -> None:
+        delivery_attempted = threading.Event()
+        release_failure = threading.Event()
+        recovery_delivered = threading.Event()
+        attempts: list[str] = []
+
+        def failed_send(
+            config: IpcConfig,
+            envelope: dict[str, object],
+            transport: dict[str, str] | None = None,
+        ) -> None:
+            del config, transport
+            attempts.append(str(envelope["capture_id"]))
+            delivery_attempted.set()
+            if envelope["capture_id"] == "failed":
+                release_failure.wait(1)
+                raise OSError("synthetic IPC failure")
+            recovery_delivered.set()
+
+        config = IpcConfig("127.0.0.1", 1, "synthetic-token")
+        with patch("codeischeap_addon.send_envelope", side_effect=failed_send):
+            emitter = IpcEmitter(config, capacity=4, retry_delay_seconds=0.05)
+            try:
+                self.assertTrue(emitter.submit({"capture_id": "failed"}, None))
+                self.assertTrue(delivery_attempted.wait(1))
+                self.assertTrue(emitter.submit({"capture_id": "queued"}, None))
+                release_failure.set()
+                self.assertTrue(emitter._delivery_unavailable.wait(1))
+
+                started = time.monotonic()
+                self.assertFalse(emitter.submit({"capture_id": "circuit-open"}, None))
+                self.assertLess(time.monotonic() - started, 0.25)
+
+                retry_deadline = time.monotonic() + 1
+                while not emitter.submit({"capture_id": "recovered"}, None):
+                    self.assertLess(time.monotonic(), retry_deadline)
+                    time.sleep(0.01)
+                self.assertTrue(recovery_delivered.wait(1))
+            finally:
+                release_failure.set()
+                emitter.close()
+
+        self.assertEqual(attempts, ["failed", "recovered"])
+
 
 if __name__ == "__main__":
     unittest.main()