test(mock): add scoped model faults and embedding recovery controls (#2790)

tools/mock-backends.md

@@ -0,0 +1,102 @@
+# Deterministic model faults
+
+Run the test backend with Python 3.11 or newer:
+
+```sh
+python tools/mock_llm.py --port 18080 --record /tmp/mock-requests.jsonl
+```
+
+Point the LangBot OpenAI-compatible provider at `http://127.0.0.1:18080/v1`
+and choose a model below. The default `mock-chat`, `mock-embedding`, tool and
+smoke models retain their existing behavior. `GET /v1/models` lists all models.
+
+| Model | Injected behavior |
+| --- | --- |
+| `mock-stream-disconnect` | Send two SSE chunks, then close an incomplete HTTP chunked response. Requires streaming. |
+| `mock-stream-stall` | Pause for two seconds after the second SSE chunk, then finish. Requires streaming. |
+| `mock-recover-503` | Return 503 on the first request in a scope, then `LANGBOT_RECOVERY_OK`. |
+| `mock-recover-stream-disconnect` | Disconnect the first stream in a scope; later streams finish with `LANGBOT_STREAM_RECOVERY_OK`. |
+| `mock-embedding-429`, `mock-embedding-503` | Return the named HTTP error. |
+| `mock-embedding-timeout` | Wait two seconds before responding. Set the client timeout below two seconds to observe a timeout. |
+| `mock-embedding-wrong-count` | Omit the final input's vector, including an empty result for one input. |
+| `mock-embedding-wrong-dimension` | Return one fewer dimension, or two dimensions when one was requested. Supports float and base64 encoding. |
+| `mock-embedding-recover-503` | Return 503 on the first embedding request in a scope, then normal vectors. |
+
+The daily tool recovery models validate returned tool results before advancing.
+`mock-daily-empty-rag-chat` rejects either configured knowledge sentinel anywhere
+in the model messages. Token-hash embeddings exercise protocol and retrieval
+plumbing; they do not evaluate semantic model quality.
+
+## Isolate recovery attempts
+
+Assign each run/case/conversation a unique scope, such as `run17.rag.group3`.
+Set this in the LangBot model's advanced parameters:
+
+```json
+{"extra_body": {"mock_scope": "run17.rag.group3"}}
+```
+
+The Core LiteLLM requester forwards this as top-level `mock_scope` in both chat
+and embedding requests. A model-level value is shared by every conversation
+using that model: use a separate model configuration per scope or use a marker
+`LB_MOCK_SCOPE=run17.rag.group3` in the current user message. Markers in older
+chat history are ignored. Embedding input strings also support markers, but
+they change the hashed vectors; prefer `extra_body` for embedding tests.
+An explicit `mock_scope` overrides markers. IDs accept 1–128 ASCII letters,
+digits and `_.:-`. Scopes are test correlation IDs, not tenant authorization.
+
+To exercise the same contract directly, run this request twice; the first
+returns 503 and the second returns the recovery message:
+
+```sh
+curl -i http://127.0.0.1:18080/v1/chat/completions \
+  -H 'Content-Type: application/json' \
+  -d '{"model":"mock-recover-503","mock_scope":"run17.direct","messages":[{"role":"user","content":"hello"}]}'
+```
+
+Counters are isolated by endpoint, model name and scope. Each validated request,
+including an SDK retry, consumes an attempt; tool steps still follow the actual
+tool history. SDK retries can consume the failure and recovery within one
+product call: the current Core embedding path retries a 503 even when model
+advanced parameters set `max_retries: 0` and `num_retries: 0`. Use the permanent
+error models to assert a visible error, and inspect recorded attempts for
+recovery scenarios. A request rejected by prompt/tool/input validation does not
+consume an attempt.
+
+After draining a scenario's requests, release all its counters:
+
+```sh
+curl -X DELETE http://127.0.0.1:18080/_mock/scopes/run17.rag.group3
+```
+
+The response reports the number released. Reusing a released scope restarts its
+fault sequence. Idle counters expire after one hour; at 2,048 retained counters,
+new scopes receive 400 until counters are released or expire. Existing scopes
+continue working. The backend is unauthenticated and defaults to loopback;
+use it only in the test environment.
+
+## Configure and verify a fault
+
+Custom `--scenarios` JSON can set `status`, `delay_seconds`,
+`chunk_delay_seconds`, `disconnect_after_chunks`, or the pair
+`stall_after_chunks` / `stall_seconds`. Delays are bounded to 0–60 seconds;
+the first role chunk counts toward stream positions. Add `recover_after: N`
+to apply the controls only to the first N requests per scope (1–100).
+Embedding scenarios use `kind: "embedding"` and optionally
+`embedding_fault: "count"` or `"dimension"`.
+
+Each completed handler appends the request, scope, attempt (for recovery), HTTP
+status, actual `faults`, and `response_complete` to the JSONL record. Delays
+include measured elapsed time. A deliberate stream cut records its chunk count
+and `injected_disconnect`; socket errors caused by the client record
+`disconnected`. `response` contains the planned chat message or embedding
+payload; `response_complete` reports whether the server finished writing it,
+not whether the client received it. Check the client result too. A configured
+stream position that was never reached produces no injection event and cannot
+count as a successful fault test. Records contain request bodies; use test data.
+
+Run the HTTP tests without installing LangBot dependencies:
+
+```sh
+python -m unittest discover -s tools -p test_mock_llm.py -v
+```

tools/mock_llm.py

@@ -9,6 +9,7 @@ import json
 import math
 from pathlib import Path
 import re
+import socket
 import struct
 import threading
 import time
@@ -18,6 +19,36 @@ import uuid
 MAX_BODY_BYTES = 1024 * 1024
 USAGE = {'prompt_tokens': 10, 'completion_tokens': 5, 'total_tokens': 15}
 GROUP_TURN = re.compile(r'LB_GROUP=(\S+) LB_TURN=(\d+) LB_TOKEN=([a-f0-9]+) LB_PREVIOUS=(none|[a-f0-9]+)')
+SCOPE_MARKER = re.compile(r'\bLB_MOCK_SCOPE=([^\s]+)')
+SCOPE_ID = re.compile(r'[A-Za-z0-9_.:-]{1,128}')
+MAX_FAULT_SCOPES = 2048
+FAULT_SCOPE_TTL_SECONDS = 3600
+
+
+class StreamDisconnected(Exception):
+    """The mock deliberately closed an unfinished HTTP chunked response."""
+
+
+def fault_scope(body, embedding):
+    if 'mock_scope' in body:
+        scope = body['mock_scope']
+    else:
+        if not embedding:
+            text = next(
+                (message.get('content', '') for message in reversed(body['messages']) if message.get('role') == 'user'),
+                '',
+            )
+        else:
+            text = body.get('input', '')
+        if isinstance(text, list):
+            text = '\n'.join(str(item.get('text', '') if isinstance(item, dict) else item) for item in text)
+        scopes = set(SCOPE_MARKER.findall(str(text)))
+        if len(scopes) > 1:
+            raise ValueError('request contains conflicting LB_MOCK_SCOPE markers')
+        scope = next(iter(scopes), None)
+    if scope is not None and (not isinstance(scope, str) or not SCOPE_ID.fullmatch(scope)):
+        raise ValueError('mock_scope must contain 1..128 letters, digits, or _.:-')
+    return scope
 
 
 def group_ack(messages, start):
@@ -116,6 +147,8 @@ def select_message(body, scenario):
     if expected_prompt := step.get('expect_prompt'):
         if expected_prompt not in json.dumps(messages, ensure_ascii=False):
             raise ValueError(f'model input must contain {expected_prompt!r}')
+    if any(value in json.dumps(messages, ensure_ascii=False) for value in step.get('forbid_prompt', [])):
+        raise ValueError('model input contains forbidden knowledge content')
     expected = step.get('expect_tool_output')
     if expected is not None:
         results = turn[calls[-1] + 1 :] if calls else []
@@ -151,10 +184,34 @@ class MockServer(ThreadingHTTPServer):
     daemon_threads = True
 
     def __init__(self, address, scenarios, record_path=None):
+        validate_scenarios(scenarios)
         super().__init__(address, Handler)
-        self.scenarios = scenarios
+        self.scenarios = {'mock-embedding': {'kind': 'embedding'}, **scenarios}
         self.record_path = record_path
         self.record_lock = threading.Lock()
+        self.fault_lock = threading.Lock()
+        self.fault_scopes = {}
+
+    def fault_controls(self, body, scenario, entry):
+        scope = fault_scope(body, entry['path'] == '/v1/embeddings')
+        entry['scope'] = scope
+        if 'recover_after' not in scenario:
+            return scenario
+        if scope is None:
+            raise ValueError('recovery scenarios require mock_scope or LB_MOCK_SCOPE')
+        key = (entry['path'], body['model'], scope)
+        now = time.monotonic()
+        with self.fault_lock:
+            for expired in [
+                key for key, (_, touched) in self.fault_scopes.items() if now - touched >= FAULT_SCOPE_TTL_SECONDS
+            ]:
+                del self.fault_scopes[expired]
+            if key not in self.fault_scopes and len(self.fault_scopes) >= MAX_FAULT_SCOPES:
+                raise ValueError('mock fault scope limit reached; release completed scopes first')
+            attempt = self.fault_scopes.get(key, (0, now))[0] + 1
+            self.fault_scopes[key] = (attempt, now)
+        entry.update(attempt=attempt, fault_active=attempt <= scenario['recover_after'])
+        return scenario if entry['fault_active'] else {}
 
     def record(self, entry):
         if self.record_path:
@@ -192,7 +249,7 @@ class Handler(BaseHTTPRequestHandler):
                             'created': 0,
                             'owned_by': 'mock',
                         }
-                        for name in [*self.server.scenarios, 'mock-embedding']
+                        for name in self.server.scenarios
                     ],
                 },
             )
@@ -207,8 +264,33 @@ class Handler(BaseHTTPRequestHandler):
                 },
             )
 
+    def do_DELETE(self):
+        prefix = '/_mock/scopes/'
+        scope = self.path.removeprefix(prefix)
+        if not self.path.startswith(prefix) or not SCOPE_ID.fullmatch(scope):
+            self.json_response(400, {'error': {'message': 'use DELETE /_mock/scopes/<scope>'}})
+            return
+        with self.server.fault_lock:
+            keys = [key for key in self.server.fault_scopes if key[2] == scope]
+            for key in keys:
+                del self.server.fault_scopes[key]
+        self.json_response(200, {'scope': scope, 'released': len(keys)})
+
+    def delay(self, seconds, kind, entry):
+        if seconds:
+            start = time.monotonic()
+            time.sleep(seconds)
+            entry['faults'].append({'kind': kind, 'seconds': seconds, 'elapsed_seconds': time.monotonic() - start})
+
     def do_POST(self):
-        entry = {'request_id': uuid.uuid4().hex, 'time': time.time(), 'path': self.path, 'status': 400}
+        entry = {
+            'request_id': uuid.uuid4().hex,
+            'time': time.time(),
+            'path': self.path,
+            'status': 400,
+            'faults': [],
+            'response_complete': False,
+        }
         try:
             if self.path not in ('/v1/chat/completions', '/v1/embeddings'):
                 entry['status'] = 404
@@ -221,6 +303,7 @@ class Handler(BaseHTTPRequestHandler):
                         }
                     },
                 )
+                entry['response_complete'] = True
                 return
             self.connection.settimeout(10)
             length = int(self.headers.get('Content-Length', 0))
@@ -230,31 +313,43 @@ class Handler(BaseHTTPRequestHandler):
             if not isinstance(body, dict) or not isinstance(body.get('model'), str):
                 raise ValueError('request must be an object with a model name')
             entry['request'] = body
-            if self.path == '/v1/embeddings':
-                if body['model'] != 'mock-embedding':
-                    raise ValueError('embedding model must be mock-embedding')
-                payload = mock_embeddings(body)
-                entry.update(status=200, response=payload)
-                self.json_response(200, payload)
-                return
             scenario = self.server.scenarios.get(body['model'])
             if scenario is None:
                 raise ValueError('unknown model; use GET /v1/models')
-            message = select_message(body, scenario)
-            time.sleep(scenario.get('delay_seconds', 0))
-            status = scenario.get('status', 200)
+            embedding = self.path == '/v1/embeddings'
+            if embedding != (scenario.get('kind') == 'embedding'):
+                raise ValueError('model does not support this endpoint')
+            if embedding:
+                payload = mock_embeddings(body)
+            else:
+                message = select_message(body, scenario)
+                if not body.get('stream') and (
+                    'disconnect_after_chunks' in scenario or 'stall_after_chunks' in scenario
+                ):
+                    raise ValueError('this fault scenario requires stream=true')
+            controls = self.server.fault_controls(body, scenario, entry)
+            self.delay(controls.get('delay_seconds', 0), 'delay', entry)
+            status = controls.get('status', 200)
             entry['status'] = status
             if status != 200:
-                self.json_response(
-                    status,
-                    {
-                        'error': {
-                            'message': 'injected mock failure',
-                            'type': 'api_error',
-                            'code': str(status),
-                        }
-                    },
-                )
+                entry['faults'].append({'kind': 'http_error', 'status': status})
+                entry['response'] = {
+                    'error': {'message': 'injected mock failure', 'type': 'api_error', 'code': str(status)}
+                }
+                self.json_response(status, entry['response'])
+                entry['response_complete'] = True
+                return
+            if embedding:
+                if fault := controls.get('embedding_fault'):
+                    if fault == 'count':
+                        payload['data'].pop()
+                    else:
+                        dimensions = body.get('dimensions', 384)
+                        payload = mock_embeddings({**body, 'dimensions': dimensions - 1 if dimensions > 1 else 2})
+                    entry['faults'].append({'kind': 'embedding_' + fault})
+                entry['response'] = payload
+                self.json_response(200, payload)
+                entry['response_complete'] = True
                 return
             entry['response'] = message
             base = {
@@ -264,7 +359,7 @@ class Handler(BaseHTTPRequestHandler):
             }
             finish = 'tool_calls' if message.get('tool_calls') else 'stop'
             if body.get('stream'):
-                self.stream_response(body, scenario, message, base, finish)
+                self.stream_response(body, controls, message, base, finish, entry)
             else:
                 self.json_response(
                     200,
@@ -275,29 +370,51 @@ class Handler(BaseHTTPRequestHandler):
                         'usage': USAGE,
                     },
                 )
+            entry['response_complete'] = True
+        except StreamDisconnected:
+            entry['injected_disconnect'] = True
         except (ValueError, KeyError, TypeError, TimeoutError) as exc:
             entry.update(status=400, error=str(exc))
             self.json_response(400, {'error': {'message': str(exc), 'type': 'invalid_request_error'}})
+            entry['response_complete'] = True
         except (BrokenPipeError, ConnectionResetError):
             entry['disconnected'] = True
         finally:
             self.server.record(entry)
 
-    def stream_response(self, body, scenario, message, base, finish):
+    def stream_response(self, body, scenario, message, base, finish, entry):
+        chunked = 'disconnect_after_chunks' in scenario
+        if chunked:
+            self.protocol_version = 'HTTP/1.1'
         self.send_response(200)
         self.send_header('Content-Type', 'text/event-stream')
         self.send_header('Cache-Control', 'no-cache')
+        if chunked:
+            self.send_header('Transfer-Encoding', 'chunked')
         self.end_headers()
 
+        def write(data):
+            if chunked:
+                data = f'{len(data):x}\r\n'.encode() + data + b'\r\n'
+            self.wfile.write(data)
+            self.wfile.flush()
+
         def send(delta, reason=None):
             payload = {
                 **base,
                 'object': 'chat.completion.chunk',
                 'choices': [{'index': 0, 'delta': delta, 'finish_reason': reason}],
             }
-            self.wfile.write(('data: ' + json.dumps(payload, ensure_ascii=False) + '\n\n').encode())
-            self.wfile.flush()
-            time.sleep(scenario.get('chunk_delay_seconds', 0))
+            write(('data: ' + json.dumps(payload, ensure_ascii=False) + '\n\n').encode())
+            entry['stream_chunks'] = entry.get('stream_chunks', 0) + 1
+            if entry['stream_chunks'] == scenario.get('disconnect_after_chunks'):
+                entry['faults'].append({'kind': 'stream_disconnect', 'after_chunks': entry['stream_chunks']})
+                self.close_connection = True
+                self.connection.shutdown(socket.SHUT_RDWR)
+                raise StreamDisconnected
+            if entry['stream_chunks'] == scenario.get('stall_after_chunks'):
+                self.delay(scenario['stall_seconds'], 'stream_stall', entry)
+            self.delay(scenario.get('chunk_delay_seconds', 0), 'chunk_delay', entry)
 
         send({'role': 'assistant'})
         content = message.get('content') or ''
@@ -339,9 +456,65 @@ class Handler(BaseHTTPRequestHandler):
                 'choices': [],
                 'usage': USAGE,
             }
-            self.wfile.write(('data: ' + json.dumps(payload) + '\n\n').encode())
-        self.wfile.write(b'data: [DONE]\n\n')
-        self.wfile.flush()
+            write(('data: ' + json.dumps(payload) + '\n\n').encode())
+        write(b'data: [DONE]\n\n')
+        if chunked:
+            self.wfile.write(b'0\r\n\r\n')
+            self.wfile.flush()
+
+
+def validate_scenarios(scenarios):
+    if not isinstance(scenarios, dict) or not scenarios:
+        raise ValueError('scenarios must be a non-empty object keyed by model name')
+    for name, scenario in scenarios.items():
+        if not isinstance(scenario, dict):
+            raise ValueError(f'{name}: scenario must be an object')
+        if scenario.get('kind', 'chat') not in ('chat', 'embedding'):
+            raise ValueError(f'{name}: kind must be chat or embedding')
+        steps = scenario.get('steps', [])
+        if not isinstance(steps, list) or (not steps and scenario.get('kind') != 'embedding'):
+            raise ValueError(f'{name}: steps must be a non-empty array')
+        if any(not isinstance(step, dict) for step in steps):
+            raise ValueError(f'{name}: each step must be an object')
+        for step in steps:
+            if not isinstance(step.get('content'), str) and not step.get('tool_calls'):
+                raise ValueError(f'{name}: each step needs content or tool_calls')
+            if 'expect_tool_output' in step and not isinstance(step['expect_tool_output'], str):
+                raise ValueError(f'{name}: expect_tool_output must be a string')
+            if 'expect_prompt' in step and not isinstance(step['expect_prompt'], str):
+                raise ValueError(f'{name}: expect_prompt must be a string')
+            if 'forbid_prompt' in step and (
+                not isinstance(step['forbid_prompt'], list)
+                or any(not isinstance(value, str) or not value for value in step['forbid_prompt'])
+            ):
+                raise ValueError(f'{name}: forbid_prompt must contain non-empty strings')
+            if 'tool_calls' in step:
+                calls = step['tool_calls']
+                if not isinstance(calls, list) or any(
+                    not isinstance(call, dict)
+                    or not isinstance(call.get('name'), str)
+                    or not isinstance(call.get('arguments'), dict)
+                    for call in calls
+                ):
+                    raise ValueError(f'{name}: tool_calls need a name and arguments object')
+        if type(scenario.get('status', 200)) is not int or scenario.get('status', 200) not in (200, 401, 429, 500, 503):
+            raise ValueError(f'{name}: unsupported injected HTTP status')
+        for key in ('delay_seconds', 'chunk_delay_seconds', 'stall_seconds'):
+            value = scenario.get(key, 0)
+            if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 <= value <= 60:
+                raise ValueError(f'{name}: {key} must be between 0 and 60')
+        for key in ('recover_after', 'disconnect_after_chunks', 'stall_after_chunks'):
+            if key in scenario and (type(scenario[key]) is not int or not 1 <= scenario[key] <= 100):
+                raise ValueError(f'{name}: {key} must be an integer in 1..100')
+        if ('stall_after_chunks' in scenario) != ('stall_seconds' in scenario):
+            raise ValueError(f'{name}: stall_after_chunks and stall_seconds must be set together')
+        if scenario.get('embedding_fault') not in (None, 'count', 'dimension'):
+            raise ValueError(f'{name}: embedding_fault must be count or dimension')
+        if scenario.get('kind') == 'embedding':
+            if any(key in scenario for key in ('disconnect_after_chunks', 'stall_after_chunks', 'chunk_delay_seconds')):
+                raise ValueError(f'{name}: embedding scenarios cannot set stream controls')
+        elif 'embedding_fault' in scenario:
+            raise ValueError(f'{name}: embedding_fault requires kind=embedding')
 
 
 def main():
@@ -360,35 +533,10 @@ def main():
     )
     args = parser.parse_args()
     scenarios = json.loads(args.scenarios.read_text(encoding='utf-8'))
-    if not isinstance(scenarios, dict) or not scenarios:
-        parser.error('scenarios must be a non-empty object keyed by model name')
-    for name, scenario in scenarios.items():
-        if not isinstance(scenario, dict) or not isinstance(scenario.get('steps'), list) or not scenario['steps']:
-            parser.error(f'{name}: steps must be a non-empty array')
-        if any(not isinstance(step, dict) for step in scenario['steps']):
-            parser.error(f'{name}: each step must be an object')
-        for step in scenario['steps']:
-            if not isinstance(step.get('content'), str) and not step.get('tool_calls'):
-                parser.error(f'{name}: each step needs content or tool_calls')
-            if 'expect_tool_output' in step and not isinstance(step['expect_tool_output'], str):
-                parser.error(f'{name}: expect_tool_output must be a string')
-            if 'expect_prompt' in step and not isinstance(step['expect_prompt'], str):
-                parser.error(f'{name}: expect_prompt must be a string')
-            if 'tool_calls' in step:
-                calls = step['tool_calls']
-                if not isinstance(calls, list) or any(
-                    not isinstance(call, dict)
-                    or not isinstance(call.get('name'), str)
-                    or not isinstance(call.get('arguments'), dict)
-                    for call in calls
-                ):
-                    parser.error(f'{name}: tool_calls need a name and arguments object')
-        if scenario.get('status', 200) not in (200, 401, 429, 500, 503):
-            parser.error(f'{name}: unsupported injected HTTP status')
-        for key in ('delay_seconds', 'chunk_delay_seconds'):
-            value = scenario.get(key, 0)
-            if not isinstance(value, (int, float)) or not 0 <= value <= 60:
-                parser.error(f'{name}: {key} must be between 0 and 60')
+    try:
+        validate_scenarios(scenarios)
+    except ValueError as exc:
+        parser.error(str(exc))
     with MockServer((args.host, args.port), scenarios, args.record) as server:
         print(f'Mock LLM: http://{args.host}:{server.server_port}/v1', flush=True)
         try:

tools/mock_llm_scenarios.json

@@ -165,5 +165,135 @@
         "content": "LANGBOT_OTHER_RAG_OK"
       }
     ]
+  },
+  "mock-daily-tool-recovery": {
+    "steps": [
+      {
+        "tool_calls": [
+          {
+            "name": "exec",
+            "arguments": {
+              "command": "printf EXPECTED_RECOVERY_ERROR >&2; exit 7"
+            }
+          }
+        ]
+      },
+      {
+        "expect_tool_output": "EXPECTED_RECOVERY_ERROR",
+        "tool_calls": [
+          {
+            "name": "exec",
+            "arguments": {
+              "command": "printf RECOVERY_OK"
+            }
+          }
+        ]
+      },
+      {
+        "expect_tool_output": "RECOVERY_OK",
+        "content": "LANGBOT_TOOL_RECOVERY_OK"
+      }
+    ]
+  },
+  "mock-daily-tool-timeout-recovery": {
+    "steps": [
+      {
+        "tool_calls": [
+          {
+            "name": "exec",
+            "arguments": {
+              "command": "rm -f /tmp/langbot-timeout-leak; sleep 3; printf LEAK > /tmp/langbot-timeout-leak",
+              "timeout_sec": 1
+            }
+          }
+        ]
+      },
+      {
+        "expect_tool_output": "timed_out",
+        "tool_calls": [
+          {
+            "name": "exec",
+            "arguments": {
+              "command": "sleep 3; test ! -e /tmp/langbot-timeout-leak && printf TIMEOUT_RECOVERY_OK"
+            }
+          }
+        ]
+      },
+      {
+        "expect_tool_output": "TIMEOUT_RECOVERY_OK",
+        "content": "LANGBOT_TOOL_TIMEOUT_RECOVERY_OK"
+      }
+    ]
+  },
+  "mock-daily-empty-rag-chat": {
+    "steps": [
+      {
+        "forbid_prompt": [
+          "aurora-parser-rag-9137",
+          "cedar-isolation-4826"
+        ],
+        "content": "LANGBOT_EMPTY_RAG_CHAT_OK"
+      }
+    ]
+  },
+  "mock-stream-disconnect": {
+    "disconnect_after_chunks": 2,
+    "steps": [
+      {
+        "content": "LANGBOT_STREAM_OK"
+      }
+    ]
+  },
+  "mock-stream-stall": {
+    "stall_after_chunks": 2,
+    "stall_seconds": 2,
+    "steps": [
+      {
+        "content": "LANGBOT_STREAM_OK"
+      }
+    ]
+  },
+  "mock-recover-503": {
+    "status": 503,
+    "recover_after": 1,
+    "steps": [
+      {
+        "content": "LANGBOT_RECOVERY_OK"
+      }
+    ]
+  },
+  "mock-recover-stream-disconnect": {
+    "disconnect_after_chunks": 2,
+    "recover_after": 1,
+    "steps": [
+      {
+        "content": "LANGBOT_STREAM_RECOVERY_OK"
+      }
+    ]
+  },
+  "mock-embedding-429": {
+    "kind": "embedding",
+    "status": 429
+  },
+  "mock-embedding-503": {
+    "kind": "embedding",
+    "status": 503
+  },
+  "mock-embedding-timeout": {
+    "kind": "embedding",
+    "delay_seconds": 2
+  },
+  "mock-embedding-wrong-count": {
+    "kind": "embedding",
+    "embedding_fault": "count"
+  },
+  "mock-embedding-wrong-dimension": {
+    "kind": "embedding",
+    "embedding_fault": "dimension"
+  },
+  "mock-embedding-recover-503": {
+    "kind": "embedding",
+    "status": 503,
+    "recover_after": 1
   }
 }

tools/test_mock_llm.py

@@ -2,16 +2,20 @@
 
 from concurrent.futures import ThreadPoolExecutor
 import base64
+from http.client import IncompleteRead
 import json
 from pathlib import Path
 import tempfile
 import struct
 import threading
+import time
 import unittest
+from unittest.mock import patch
 from urllib.error import HTTPError
 from urllib.request import Request, urlopen
 
-from mock_llm import MockServer
+import mock_llm
+from mock_llm import MockServer, validate_scenarios
 
 
 class MockLLMTest(unittest.TestCase):
@@ -19,7 +23,18 @@ class MockLLMTest(unittest.TestCase):
     def setUpClass(cls):
         cls.tmp = tempfile.TemporaryDirectory()
         cls.record = Path(cls.tmp.name) / 'requests.jsonl'
+        cls.record.touch()
         scenarios = json.loads(Path(__file__).with_name('mock_llm_scenarios.json').read_text())
+        scenarios['test-stall-recovery'] = {
+            **scenarios['mock-stream-stall'],
+            'stall_seconds': 0.2,
+            'recover_after': 1,
+        }
+        scenarios['test-delay-recovery'] = {
+            'delay_seconds': 0.2,
+            'recover_after': 1,
+            'steps': [{'content': 'RECOVERED'}],
+        }
         cls.server = MockServer(('127.0.0.1', 0), scenarios, cls.record)
         cls.thread = threading.Thread(target=cls.server.serve_forever)
         cls.thread.start()
@@ -32,7 +47,7 @@ class MockLLMTest(unittest.TestCase):
         cls.thread.join()
         cls.tmp.cleanup()
 
-    def post(self, model='mock-chat', **kwargs):
+    def post(self, model='mock-chat', timeout=5, **kwargs):
         body = {
             'model': model,
             'messages': [{'role': 'user', 'content': 'test'}],
@@ -46,9 +61,34 @@ class MockLLMTest(unittest.TestCase):
                 'Authorization': 'Bearer must-not-be-recorded',
             },
         )
-        with urlopen(request, timeout=5) as response:
+        with urlopen(request, timeout=timeout) as response:
             return response.read().decode()
 
+    def embed(self, timeout=5, **patch):
+        body = {'model': 'mock-embedding', 'input': ['repeatable fixture', 'another fixture'], **patch}
+        request = Request(
+            self.url + '/v1/embeddings',
+            data=json.dumps(body).encode(),
+            headers={'Content-Type': 'application/json'},
+        )
+        with urlopen(request, timeout=timeout) as response:
+            return json.load(response)
+
+    def records(self, scope, count):
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            with self.server.record_lock:
+                rows = [json.loads(line) for line in self.record.read_text().splitlines()]
+            rows = [row for row in rows if row.get('scope') == scope]
+            if len(rows) >= count:
+                return sorted(rows, key=lambda row: row.get('attempt', 0))
+            time.sleep(0.01)
+        self.fail(f'missing completed request records for {scope}')
+
+    def release(self, scope):
+        with urlopen(Request(self.url + '/_mock/scopes/' + scope, method='DELETE'), timeout=5) as response:
+            return json.load(response)
+
     def test_chat_and_model_discovery(self):
         with urlopen(self.url + '/v1/models') as response:
             models = json.load(response)
@@ -80,28 +120,18 @@ class MockLLMTest(unittest.TestCase):
             self.assertEqual(failure.exception.code, 400)
 
     def test_embedding_http_float_base64_and_validation(self):
-        def embed(**patch):
-            body = {'model': 'mock-embedding', 'input': ['repeatable fixture', 'another fixture'], **patch}
-            request = Request(
-                self.url + '/v1/embeddings',
-                data=json.dumps(body).encode(),
-                headers={'Content-Type': 'application/json'},
-            )
-            with urlopen(request, timeout=5) as response:
-                return json.load(response)
-
-        first = embed()['data']
-        self.assertEqual(first, embed()['data'])
+        first = self.embed()['data']
+        self.assertEqual(first, self.embed()['data'])
         self.assertEqual([item['index'] for item in first], [0, 1])
         self.assertEqual(len(first[0]['embedding']), 384)
         self.assertAlmostEqual(sum(value * value for value in first[0]['embedding']), 1)
-        encoded = embed(encoding_format='base64')['data'][0]['embedding']
+        encoded = self.embed(encoding_format='base64')['data'][0]['embedding']
         decoded = struct.unpack('<384f', base64.b64decode(encoded))
         for actual, expected in zip(decoded, first[0]['embedding']):
             self.assertAlmostEqual(actual, expected, places=6)
         for patch in ({'input': []}, {'dimensions': 0}, {'encoding_format': 'bad'}, {'model': 'unknown'}):
             with self.subTest(patch=patch), self.assertRaises(HTTPError) as caught:
-                embed(**patch)
+                self.embed(**patch)
             self.assertEqual(caught.exception.code, 400)
 
     def box_turn(self, streaming=False):
@@ -171,6 +201,185 @@ class MockLLMTest(unittest.TestCase):
         )
         self.assertEqual(response['choices'][0]['message']['content'], 'LANGBOT_RAG_OK')
 
+    def test_empty_rag_rejects_context_from_another_knowledge_base(self):
+        self.assertIn('LANGBOT_EMPTY_RAG_CHAT_OK', self.post('mock-daily-empty-rag-chat'))
+        for leaked in ('aurora-parser-rag-9137', 'cedar-isolation-4826'):
+            with self.subTest(leaked=leaked), self.assertRaises(HTTPError) as caught:
+                self.post('mock-daily-empty-rag-chat', messages=[{'role': 'user', 'content': leaked}])
+            self.assertEqual(caught.exception.code, 400)
+
+    def test_concurrent_scopes_fail_independently_then_recover(self):
+        def conversation(index):
+            scope = f'concurrent-{index}'
+            body = {'messages': [{'role': 'user', 'content': f'hello LB_MOCK_SCOPE={scope}'}]}
+            with self.assertRaises(HTTPError) as caught:
+                self.post('mock-recover-503', **body)
+            self.assertEqual(caught.exception.code, 503)
+            self.assertIn('LANGBOT_CHAT_OK', self.post())
+            self.assertIn('LANGBOT_RECOVERY_OK', self.post('mock-recover-503', **body))
+            rows = self.records(scope, 2)
+            self.assertEqual([row['attempt'] for row in rows], [1, 2])
+            self.assertEqual(rows[0]['faults'], [{'kind': 'http_error', 'status': 503}])
+            self.assertEqual(rows[1]['faults'], [])
+            self.assertTrue(rows[1]['response_complete'])
+            self.assertEqual(self.release(scope)['released'], 1)
+
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            list(pool.map(conversation, range(8)))
+
+    def test_concurrent_retries_consume_only_one_failure(self):
+        def request(_):
+            try:
+                self.post('mock-recover-503', mock_scope='same-scope')
+                return 200
+            except HTTPError as exc:
+                return exc.code
+
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            statuses = list(pool.map(request, range(8)))
+        self.assertEqual(sorted(statuses), [200] * 7 + [503])
+        rows = self.records('same-scope', 8)
+        self.assertEqual(sorted(row['attempt'] for row in rows), list(range(1, 9)))
+        self.assertEqual(self.release('same-scope')['released'], 1)
+
+    def test_stream_disconnect_is_a_transport_failure_and_recovers(self):
+        with self.assertRaises(IncompleteRead) as caught:
+            self.post('mock-recover-stream-disconnect', stream=True, mock_scope='disconnect')
+        self.assertIn(b'LANGBOT_', caught.exception.partial)
+        self.assertNotIn(b'[DONE]', caught.exception.partial)
+        result = self.post('mock-recover-stream-disconnect', stream=True, mock_scope='disconnect')
+        self.assertIn('data: [DONE]', result)
+        rows = self.records('disconnect', 2)
+        self.assertEqual(rows[0]['faults'], [{'kind': 'stream_disconnect', 'after_chunks': 2}])
+        self.assertFalse(rows[0]['response_complete'])
+        self.assertTrue(rows[0]['injected_disconnect'])
+        self.assertEqual(rows[1]['faults'], [])
+        self.assertTrue(rows[1]['response_complete'])
+        self.release('disconnect')
+
+    def test_delay_and_midstream_stall_timeout_then_recover(self):
+        for model, stream, kind in (
+            ('test-delay-recovery', False, 'delay'),
+            ('test-stall-recovery', True, 'stream_stall'),
+        ):
+            with self.subTest(model=model):
+                with self.assertRaises(TimeoutError):
+                    self.post(model, stream=stream, timeout=0.05, mock_scope=model)
+                self.assertIn('LANGBOT_CHAT_OK', self.post())
+                result = self.post(model, stream=stream, mock_scope=model)
+                self.assertIn('data: [DONE]' if stream else 'RECOVERED', result)
+                rows = sorted(self.records(model, 2), key=lambda row: row['attempt'])
+                self.assertEqual(rows[0]['faults'][0]['kind'], kind)
+                self.assertGreaterEqual(rows[0]['faults'][0]['elapsed_seconds'], 0.2)
+                self.assertEqual(rows[1]['faults'], [])
+                self.release(model)
+
+    def test_embedding_errors_timeout_count_and_dimension(self):
+        for status in (429, 503):
+            scope = f'embed-{status}'
+            with self.assertRaises(HTTPError) as caught:
+                self.embed(model=f'mock-embedding-{status}', mock_scope=scope)
+            self.assertEqual(caught.exception.code, status)
+            self.assertEqual(self.records(scope, 1)[0]['faults'], [{'kind': 'http_error', 'status': status}])
+        with self.assertRaises(TimeoutError):
+            self.embed(model='mock-embedding-timeout', mock_scope='embed-timeout', timeout=0.05)
+        self.assertEqual(len(self.embed()['data']), 2)
+        for encoding in ('float', 'base64'):
+            wrong_count = self.embed(model='mock-embedding-wrong-count', encoding_format=encoding)
+            self.assertEqual(len(wrong_count['data']), 1)
+            wrong_dimension = self.embed(
+                model='mock-embedding-wrong-dimension',
+                dimensions=4,
+                encoding_format=encoding,
+            )
+            self.assertEqual([item['index'] for item in wrong_dimension['data']], [0, 1])
+            vector = wrong_dimension['data'][0]['embedding']
+            self.assertEqual(len(vector) if encoding == 'float' else len(base64.b64decode(vector)) // 4, 3)
+        delay = self.records('embed-timeout', 1)[0]['faults'][0]
+        self.assertEqual(delay['kind'], 'delay')
+        self.assertGreaterEqual(delay['elapsed_seconds'], 2)
+
+    def test_embedding_recovery_scope_does_not_modify_vectors(self):
+        expected = self.embed()['data']
+
+        def conversation(scope):
+            with self.assertRaises(HTTPError) as caught:
+                self.embed(model='mock-embedding-recover-503', mock_scope=scope)
+            self.assertEqual(caught.exception.code, 503)
+            result = self.embed(model='mock-embedding-recover-503', mock_scope=scope)
+            self.assertEqual(result['data'], expected)
+            self.assertEqual(self.records(scope, 2)[1]['faults'], [])
+            self.release(scope)
+
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            list(pool.map(conversation, ('embed-a', 'embed-b')))
+
+    def test_markers_select_current_scope_and_invalid_input_does_not_advance(self):
+        with self.assertRaises(HTTPError) as caught:
+            self.post('mock-recover-503', messages=[], mock_scope='marker-current')
+        self.assertEqual(caught.exception.code, 400)
+        with self.assertRaises(HTTPError) as caught:
+            self.post(
+                'mock-recover-503',
+                messages=[
+                    {'role': 'user', 'content': 'LB_MOCK_SCOPE=marker-old'},
+                    {'role': 'assistant', 'content': 'prior reply'},
+                    {'role': 'user', 'content': [{'type': 'text', 'text': 'LB_MOCK_SCOPE=marker-current'}]},
+                ],
+            )
+        self.assertEqual(caught.exception.code, 503)
+        self.assertEqual(self.records('marker-current', 1)[0]['attempt'], 1)
+        self.assertIn('LANGBOT_RECOVERY_OK', self.post('mock-recover-503', mock_scope='marker-current'))
+        self.assertEqual(self.release('marker-current')['released'], 1)
+        self.assertEqual(self.release('marker-old')['released'], 0)
+        with self.assertRaises(HTTPError) as caught:
+            self.embed(model='mock-embedding-recover-503', input=['LB_MOCK_SCOPE=marker-embed'])
+        self.assertEqual(caught.exception.code, 503)
+        self.assertEqual(self.records('marker-embed', 1)[0]['attempt'], 1)
+        self.release('marker-embed')
+
+    def test_scope_validation_capacity_expiry_and_cleanup(self):
+        for scope in (None, '', [], 'x' * 129, 'bad/scope'):
+            with self.subTest(scope=scope), self.assertRaises(HTTPError) as caught:
+                self.post('mock-recover-503', mock_scope=scope)
+            self.assertEqual(caught.exception.code, 400)
+        with patch.object(mock_llm, 'MAX_FAULT_SCOPES', len(self.server.fault_scopes) + 1):
+            with self.assertRaises(HTTPError) as caught:
+                self.post('mock-recover-503', mock_scope='capacity-a')
+            self.assertEqual(caught.exception.code, 503)
+            with self.assertRaises(HTTPError) as caught:
+                self.post('mock-recover-503', mock_scope='capacity-b')
+            self.assertEqual(caught.exception.code, 400)
+            self.assertIn('scope limit', caught.exception.read().decode())
+            self.assertIn('LANGBOT_RECOVERY_OK', self.post('mock-recover-503', mock_scope='capacity-a'))
+            self.assertEqual(self.release('capacity-a')['released'], 1)
+            with self.assertRaises(HTTPError) as caught:
+                self.post('mock-recover-503', mock_scope='capacity-b')
+            self.assertEqual(caught.exception.code, 503)
+            with patch.object(mock_llm, 'FAULT_SCOPE_TTL_SECONDS', 0), self.assertRaises(HTTPError) as caught:
+                self.post('mock-recover-503', mock_scope='capacity-b')
+            self.assertEqual(caught.exception.code, 503)
+        self.assertEqual(self.release('capacity-b')['released'], 1)
+        self.assertEqual(self.release('capacity-b')['released'], 0)
+
+    def test_scenario_controls_reject_invalid_values(self):
+        for control in (
+            {'delay_seconds': 61},
+            {'delay_seconds': float('nan')},
+            {'delay_seconds': True},
+            {'recover_after': 0},
+            {'disconnect_after_chunks': -1},
+            {'stall_after_chunks': 2},
+            {'stall_seconds': 1},
+            {'embedding_fault': 'count'},
+            {'forbid_prompt': 'not-an-array'},
+        ):
+            scenario = {'steps': [{'content': 'OK'}], **control}
+            if 'forbid_prompt' in control:
+                scenario = {'steps': [{'content': 'OK', **control}]}
+            with self.subTest(control=control), self.assertRaises(ValueError):
+                validate_scenarios({'invalid': scenario})
+
     def test_errors_and_invalid_requests(self):
         for status in (401, 429, 503):
             with self.subTest(status=status), self.assertRaises(HTTPError) as caught: