test(tools): simulate persistent multi-group conversations

tools/README.md

@@ -1,4 +1,4 @@
-# Mock LLM for regression tests
+# LangBot regression tools
 
 Run the deterministic OpenAI Chat Completions endpoint with Python 3.11+, without
 API credentials or third-party dependencies:
@@ -32,6 +32,9 @@ tool result must contain `LANGBOT_BOX_SENTINEL` before the next step is returned
 Tool-call IDs and fragmented streaming arguments use the normal OpenAI protocol.
 Scenario progress comes from each request's current conversation turn, so
 concurrent sessions and retries do not consume a shared script counter.
+Tagged group traffic additionally requires the previous user turn and assistant
+acknowledgement in the model input, and rejects another group's history. Final
+responses carry a per-turn acknowledgement checked by the traffic driver.
 
 Add fixed response/tool sequences with `--scenarios path/to/scenarios.json`, using
 `mock_llm_scenarios.json` as the example. `expect_tool_output` checks a substring
@@ -97,8 +100,10 @@ any failed health probe or process/container resource read.
 
 [`regression_suite.py`](regression_suite.py) turns a JSON profile into a repeatable
 resource preparation, traffic, observation, verdict, and cleanup sequence. Start
-with [`regression_profiles/normal.json`](regression_profiles/normal.json), which
-defines the chat/Box/RAG mix from the ten-minute trial.
+with [`regression_profiles/groups.json`](regression_profiles/groups.json) for
+continuous group conversations. The earlier
+[`normal.json`](regression_profiles/normal.json) profile retains independent CLI
+requests; parallel CLI workers alone do not simulate persistent groups.
 
 Use a dedicated QA Workspace. First provision the instance with `lbctl install`
 or your source deployment, enable Docker Box, start the mock and embedding
@@ -110,9 +115,9 @@ paths resolve against the profile directory, so adjust them if you move it.
 Run on the Linux test host:
 
 ```sh
-python3 tools/regression_suite.py validate --profile tools/regression_profiles/normal.json
+python3 tools/regression_suite.py validate --profile tools/regression_profiles/groups.json
 python3 tools/regression_suite.py run \
-  --profile tools/regression_profiles/normal.json --output .qa/regression/run-001
+  --profile tools/regression_profiles/groups.json --output .qa/regression/run-001
 ```
 
 `validate` checks profile syntax and references without contacting the instance.
@@ -124,12 +129,13 @@ The profile controls:
 
 - `plugins`: local archives or GitHub/marketplace request files, installed with
   `lbctl plugin install ... --wait`.
-- `resources`: Providers, models, knowledge bases, document ingestion, and
-  Pipelines, created in dependency order. `${resource_id}` refers to a UUID
+- `resources`: Providers, models, knowledge bases, document ingestion, Pipelines,
+  and Bots, created in dependency order. `${resource_id}` refers to a UUID
   returned by an earlier creation; `${workspace_uuid}` is the current Workspace.
-  Pipeline config patches merge with actual server defaults. Existing resources
-  can instead be supplied through `vars`, leaving `resources` empty.
-- `cases`: CLI argument arrays and expected business text. Integer `weight`
+  Pipeline config patches merge with actual server defaults. Independent CLI
+  cases can use existing resources supplied through `vars`; group mode creates
+  its own signed HTTP Bots.
+- `cases`: CLI argument arrays or `http_group` settings and expected business text. Integer `weight`
   values set the mix; `2/1/1/1` means `40/20/20/20`. An optional `body` object
   becomes the case's request file. All workers share the weighted sequence.
 - `phases`: each phase has a name, duration, target RPS, worker limit, and optional
@@ -137,6 +143,10 @@ The profile controls:
   end; the concurrency limit can reduce achieved RPS.
 - `observe`: health URLs, host PIDs, PID files, and Docker container names/IDs.
   Set PIDs for disjoint process trees to avoid counting descendants twice.
+  `box_data_root` discovers Box containers mounted below that deployment's data
+  directory on every sample, including sandboxes created as new groups speak.
+  The complete Job sets it automatically; on an existing instance, set it to
+  the instance's Box data directory when sampling dynamic Docker sandboxes.
 - `thresholds`: minimum achieved/target RPS ratio and maximum P95 seconds per
   case, in addition to the existing business and health/resource checks. The
   sample limits are starting acceptance settings, not established product SLOs.
@@ -150,7 +160,7 @@ same profile and credentials:
 
 ```sh
 python3 tools/regression_suite.py cleanup \
-  --profile tools/regression_profiles/normal.json --output .qa/regression/run-001
+  --profile tools/regression_profiles/groups.json --output .qa/regression/run-001
 ```
 
 Cleanup checks the Workspace against `owned.json` and records remaining items
@@ -161,14 +171,62 @@ target/achieved RPS, phase results, threshold failures, and the final verdict.
 
 The `regression_suite.py` entry reuses an existing instance and backend processes.
 Use the complete lifecycle Job below to own deployment, credentials and backends.
-Cases still use the existing text oracle and new-session CLI entry; continuous
-conversations, IM adapter acceptance, and resource-leak verdicts need separate
-scenarios. Test the suite plumbing on Linux with:
+The independent `normal.json` cases use a new-session CLI entry. Group traffic
+uses the signed HTTP Bot adapter instead, as described below. Real IM adapter
+acceptance and resource-leak verdicts need separate scenarios. Test the suite
+plumbing on Linux with:
 
 ```sh
 python3 -m unittest discover -s tools -p test_regression_suite.py -v
 ```
 
+### Persistent group traffic
+
+The group profile creates three HTTP Bots routed to chat, Box, and RAG Pipelines.
+Their signed `/bots/<uuid>/sync` ingress produces real `GroupMessage` events and
+uses Core's session manager. Four groups follow a `2/1/1` case assignment;
+members take turns within each group. The same member IDs appear in different
+groups to exercise group isolation. Box session selection uses the group launcher ID.
+
+In each group phase, `groups` selects the number of stable sessions and
+`members_per_group` selects the rotating member identities. `rps` remains the
+aggregate target, so a group's target speaking interval is `groups / rps`
+seconds. `workers` bounds total in-flight requests independently of group count.
+For example, 8 groups at 4 RPS target one message every 2 seconds per group.
+Both the Suite and Job accept `--groups` and `--members-per-group` overrides.
+The group count must cover the full weighted case sequence (4 in the sample).
+
+Only one request is in flight per group. A failure or unknown outcome stops that
+group without retry while other groups continue. Every selected group must
+complete at least two turns in total for the phase to pass. The fake model checks
+that the preceding user message and assistant acknowledgement survived in actual
+model input, including when a different member speaks next. Cross-group markers
+fail the oracle. The normal Box/tool-output and RAG-context checks still apply.
+
+`group-state.json` carries group IDs, turn numbers and acknowledgements across
+phases; existing groups keep their conversations when later phases add groups.
+It refuses changed cases or a state left in progress by interruption. Do not
+clear that marker to blindly resume: inspect the run and use a fresh output
+directory when the last turn's outcome is unknown. Each phase summary includes
+group-level turn counts and member IDs alongside case latency and health data.
+
+The Suite generates the HTTP Bot signing secret and passes it through
+`LANGBOT_QA_GROUP_SECRET`. Resource request bodies containing credentials are
+stored under `.private/`; do not share that directory. Cases specify `http_group`
+with `url`, `secret_env`, and `message` instead of CLI `args`. Core needs no code
+changes. The `/sync` response collapses streaming parts: this tests multi-turn
+group processing, not real IM delivery or callback streaming order.
+Every selected group must also receive traffic in each phase. Group latency
+measures the signed HTTP request through its completed reply; independent CLI
+latency includes CLI startup. Compare results within the same traffic mode.
+
+`lbctl` v0.3 may report Bot verification failure after a successful creation
+because its safe projection omits event bindings. The Suite records the returned
+Bot UUID before reading the full Core Bot API with the endpoint, Workspace and
+environment credential confirmed by `lbctl whoami`. It then
+checks the masked signing secret and effective route with a signed warmup turn
+before load. The creation is not retried.
+
 ## Complete lifecycle Job
 
 [`regression_job.py`](regression_job.py) owns the deployment as well as the suite.

tools/mock_llm.py

@@ -17,6 +17,39 @@ 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]+)')
+
+
+def group_ack(messages, start):
+    """Require a real previous user turn and reply, with no other group's history."""
+    turns = []
+    for index, message in enumerate(messages):
+        if message.get('role') == 'user':
+            for match in GROUP_TURN.finditer(str(message.get('content', ''))):
+                turns.append((index, *match.groups()))
+    if not turns:
+        return ''
+    position, group, turn, token, previous = turns[-1]
+    if position != start or any(item[1] != group for item in turns):
+        raise ValueError('group history contains another group or lacks the current turn')
+    if int(turn) == 1:
+        if len(turns) != 1 or previous != 'none':
+            raise ValueError('first group turn must have an empty history')
+    else:
+        if len(turns) < 2:
+            raise ValueError('previous group turn is missing from model history')
+        prior_position, _, prior_turn, prior_token, _ = turns[-2]
+        replies = messages[prior_position + 1 : start]
+        if (
+            int(prior_turn) != int(turn) - 1
+            or prior_token != previous
+            or not any(
+                message.get('role') == 'assistant' and f'GROUP_ACK={previous}' in str(message.get('content', ''))
+                for message in replies
+            )
+        ):
+            raise ValueError('previous group turn or assistant acknowledgement does not match')
+    return f' GROUP_ACK={token}'
 
 
 def mock_embeddings(body):
@@ -67,6 +100,7 @@ def select_message(body, scenario):
     )
     if start is None:
         raise ValueError('a user message is required')
+    acknowledgement = group_ack(messages, start)
     turn = messages[start:]
     calls = [i for i, m in enumerate(turn) if m.get('role') == 'assistant' and m.get('tool_calls')]
     for index, position in enumerate(calls):
@@ -88,7 +122,8 @@ def select_message(body, scenario):
         result_text = '\n'.join(str(m.get('content', '')) for m in results if m.get('role') == 'tool')
         if expected not in result_text:
             raise ValueError(f'tool result must contain {expected!r}')
-    message = {'role': 'assistant', 'content': step.get('content')}
+    content = step.get('content')
+    message = {'role': 'assistant', 'content': content + acknowledgement if content is not None else None}
     if tool_calls := step.get('tool_calls'):
         tools = body.get('tools', [])
         if not isinstance(tools, list) or any(

tools/regression_job.py

@@ -17,6 +17,7 @@ import urllib.request
 import yaml
 
 from regression_suite import Suite, number, resolve, validate
+from regression_traffic import owned_box_containers
 
 
 KEY_ENV = 'LANGBOT_QA_API_KEY'
@@ -222,6 +223,7 @@ class Job:
             raise RuntimeError('Docker daemon has existing Box sandboxes; use an idle dedicated test daemon')
         key = secrets.token_urlsafe(32)
         self.env[KEY_ENV] = key
+        self.env['LANGBOT_API_KEY'] = key
         self.redactions.append(key)
         (self.private / 'api-key').write_text(key)
         (self.private / 'api-key').chmod(0o600)
@@ -301,6 +303,7 @@ class Job:
         )
         self.save()
         profile = self.suite_profile
+        profile.setdefault('vars', {})['core_base_url'] = endpoint
         profile['cli'] = {'executable': self.cli[0], 'config': str(self.cli_config)}
         profile['observe'] = {'health': health, 'pids': {}, 'containers': list(containers.values())}
         for service, container in containers.items():
@@ -309,15 +312,18 @@ class Job:
         destination = self.output / 'suite'
         destination.mkdir(mode=0o700)
         suite = Suite(profile, self.suite_path, destination, env=self.env)
+        self.redactions.extend(suite.secrets)
         suite.write('profile.json', profile)
         suite.prepare()
         # Prime every configured business case before measuring steady load.
         for case in profile['cases']:
             case = resolve(case, suite.variables)
+            if 'http_group' in case:
+                continue  # Suite preparation already verifies signed group ingress and the masked secret.
             data = suite.call(case['args'], case.get('body'))
             if case['expect'] not in json.dumps(data, ensure_ascii=False):
                 raise RuntimeError(f'warmup oracle failed: {case["name"]}')
-        profile['observe']['containers'] += self.owned_sandboxes()
+        profile['observe']['box_data_root'] = str(Path(self.state['record']['data_dir']) / 'box')
         suite.write('profile.json', profile)
         return suite
 
@@ -325,16 +331,7 @@ class Job:
         record = self.state.get('record')
         if not record:
             return []
-        root = (Path(record['data_dir']) / 'box').resolve()
-        candidates = self.checked(
-            ['docker', 'ps', '-a', '--filter', 'label=langbot.box=true', '--format', '{{.ID}}']
-        ).splitlines()
-        owned = []
-        for container in candidates:
-            mounts = json.loads(self.checked(['docker', 'inspect', '--format', '{{json .Mounts}}', container]))
-            if any(Path(mount['Source']).resolve().is_relative_to(root) for mount in mounts if mount.get('Source')):
-                owned.append(container)
-        return owned
+        return owned_box_containers(Path(record['data_dir']) / 'box', self.checked)
 
     def collect(self):
         self.recover_record()
@@ -424,6 +421,8 @@ def main():
     parser.add_argument('--duration', type=float)
     parser.add_argument('--rps', type=float)
     parser.add_argument('--workers', type=int)
+    parser.add_argument('--groups', type=int)
+    parser.add_argument('--members-per-group', type=int)
     args = parser.parse_args()
     try:
         if args.action == 'cleanup':
@@ -435,6 +434,7 @@ def main():
             if (job.private / 'api-key').exists():
                 key = (job.private / 'api-key').read_text()
                 job.env[KEY_ENV] = key
+                job.env['LANGBOT_API_KEY'] = key
                 job.redactions.append(key)
             job.cleanup()
             return
@@ -442,7 +442,13 @@ def main():
             raise ValueError('--profile is required for run/validate')
         profile, suite, path = load_profile(args.profile.resolve())
         for phase in suite['phases']:
-            for flag, field in [('duration', 'duration_seconds'), ('rps', 'rps'), ('workers', 'workers')]:
+            for flag, field in [
+                ('duration', 'duration_seconds'),
+                ('rps', 'rps'),
+                ('workers', 'workers'),
+                ('groups', 'groups'),
+                ('members_per_group', 'members_per_group'),
+            ]:
                 if getattr(args, flag) is not None:
                     phase[field] = getattr(args, flag)
         validate(suite)

tools/regression_profiles/groups.json

@@ -0,0 +1,464 @@
+{
+  "version": 1,
+  "cli": {
+    "executable": "lbctl",
+    "config": "./lbctl.yaml"
+  },
+  "vars": {
+    "plugin_dir": "/path/to/plugin-zips",
+    "mock_base_url": "http://127.0.0.1:18080/v1",
+    "embedding_base_url": "http://127.0.0.1:15410/v1",
+    "core_health": "http://127.0.0.1:5300/healthz",
+    "core_base_url": "http://127.0.0.1:5300"
+  },
+  "plugins": [
+    {
+      "source": "local",
+      "file": "${plugin_dir}/LocalAgent.zip"
+    },
+    {
+      "source": "local",
+      "file": "${plugin_dir}/LangRAG.zip"
+    },
+    {
+      "source": "local",
+      "file": "${plugin_dir}/GeneralParsers.zip"
+    }
+  ],
+  "resources": [
+    {
+      "id": "mock_provider",
+      "kind": "provider",
+      "body": {
+        "name": "regression mock",
+        "requester": "openai-chat-completions",
+        "base_url": "${mock_base_url}",
+        "api_keys": [
+          "test-only"
+        ]
+      }
+    },
+    {
+      "id": "embedding_provider",
+      "kind": "provider",
+      "body": {
+        "name": "regression embedding",
+        "requester": "openai-chat-completions",
+        "base_url": "${embedding_base_url}",
+        "api_keys": [
+          "test-only"
+        ]
+      }
+    },
+    {
+      "id": "chat_model",
+      "kind": "model",
+      "type": "llm",
+      "body": {
+        "name": "mock-chat",
+        "provider_uuid": "${mock_provider}",
+        "abilities": [
+          "func_call"
+        ],
+        "context_length": 8192,
+        "extra_args": {
+          "request_policy": {
+            "retries": 0
+          }
+        }
+      }
+    },
+    {
+      "id": "box_model",
+      "kind": "model",
+      "type": "llm",
+      "body": {
+        "name": "mock-box",
+        "provider_uuid": "${mock_provider}",
+        "abilities": [
+          "func_call"
+        ],
+        "context_length": 8192,
+        "extra_args": {
+          "request_policy": {
+            "retries": 0
+          }
+        }
+      }
+    },
+    {
+      "id": "rag_model",
+      "kind": "model",
+      "type": "llm",
+      "body": {
+        "name": "mock-rag",
+        "provider_uuid": "${mock_provider}",
+        "abilities": [
+          "func_call"
+        ],
+        "context_length": 8192,
+        "extra_args": {
+          "request_policy": {
+            "retries": 0
+          }
+        }
+      }
+    },
+    {
+      "id": "embedding_model",
+      "kind": "model",
+      "type": "embedding",
+      "body": {
+        "name": "all-MiniLM-L6-v2",
+        "provider_uuid": "${embedding_provider}"
+      }
+    },
+    {
+      "id": "knowledge_base",
+      "kind": "knowledge-base",
+      "body": {
+        "name": "regression parser golden",
+        "knowledge_engine_plugin_id": "langbot-team/LangRAG",
+        "creation_settings": {
+          "embedding_model_uuid": "${embedding_model}",
+          "index_type": "chunk",
+          "chunk_size": 512,
+          "overlap": 50
+        },
+        "retrieval_settings": {
+          "search_type": "vector",
+          "top_k": 5,
+          "query_rewrite": "off",
+          "rerank": "off"
+        }
+      },
+      "documents": [
+        {
+          "file": "../../skills/skills/langbot-testing/fixtures/rag/parser-golden.html",
+          "parser_plugin_id": "langbot-team/GeneralParsers"
+        }
+      ]
+    },
+    {
+      "id": "chat_pipeline",
+      "kind": "pipeline",
+      "body": {
+        "name": "regression chat",
+        "description": "Normal deterministic chat",
+        "config": {
+          "ai": {
+            "runner": {
+              "id": "plugin:langbot-team/LocalAgent/default",
+              "expire-time": 0
+            },
+            "runner_config": {
+              "plugin:langbot-team/LocalAgent/default": {
+                "model": {
+                  "primary": "${chat_model}",
+                  "fallbacks": []
+                },
+                "prompt": [
+                  {
+                    "role": "system",
+                    "content": "You are a QA regression assistant."
+                  }
+                ],
+                "knowledge-bases": [],
+                "box-enabled": false,
+                "timeout": 30,
+                "max-round": 20,
+                "box-session-id-template": "{launcher_type}_{launcher_id}"
+              }
+            }
+          },
+          "safety": {
+            "rate-limit": {
+              "limitation": 10000
+            }
+          },
+          "output": {
+            "force-delay": {
+              "min": 0,
+              "max": 0
+            },
+            "misc": {
+              "at-sender": false,
+              "quote-origin": false
+            }
+          },
+          "trigger": {
+            "group-respond-rules": {
+              "prefix": [
+                "ai "
+              ]
+            }
+          }
+        }
+      }
+    },
+    {
+      "id": "box_pipeline",
+      "kind": "pipeline",
+      "body": {
+        "name": "regression Box",
+        "description": "Real Box write and read",
+        "config": {
+          "ai": {
+            "runner": {
+              "id": "plugin:langbot-team/LocalAgent/default",
+              "expire-time": 0
+            },
+            "runner_config": {
+              "plugin:langbot-team/LocalAgent/default": {
+                "model": {
+                  "primary": "${box_model}",
+                  "fallbacks": []
+                },
+                "prompt": [
+                  {
+                    "role": "system",
+                    "content": "You are a QA regression assistant."
+                  }
+                ],
+                "knowledge-bases": [],
+                "box-enabled": true,
+                "box-session-id-template": "{launcher_type}_{launcher_id}",
+                "timeout": 30,
+                "max-tool-iterations": 5,
+                "max-round": 20
+              }
+            }
+          },
+          "safety": {
+            "rate-limit": {
+              "limitation": 10000
+            }
+          },
+          "output": {
+            "force-delay": {
+              "min": 0,
+              "max": 0
+            },
+            "misc": {
+              "at-sender": false,
+              "quote-origin": false
+            }
+          },
+          "trigger": {
+            "group-respond-rules": {
+              "prefix": [
+                "ai "
+              ]
+            }
+          }
+        }
+      }
+    },
+    {
+      "id": "rag_pipeline",
+      "kind": "pipeline",
+      "body": {
+        "name": "regression RAG",
+        "description": "Retrieved model input oracle",
+        "config": {
+          "ai": {
+            "runner": {
+              "id": "plugin:langbot-team/LocalAgent/default",
+              "expire-time": 0
+            },
+            "runner_config": {
+              "plugin:langbot-team/LocalAgent/default": {
+                "model": {
+                  "primary": "${rag_model}",
+                  "fallbacks": []
+                },
+                "prompt": [
+                  {
+                    "role": "system",
+                    "content": "You are a QA regression assistant."
+                  }
+                ],
+                "knowledge-bases": [
+                  "${knowledge_base}"
+                ],
+                "box-enabled": false,
+                "timeout": 30,
+                "retrieval-top-k": 5,
+                "max-round": 20,
+                "box-session-id-template": "{launcher_type}_{launcher_id}"
+              }
+            }
+          },
+          "safety": {
+            "rate-limit": {
+              "limitation": 10000
+            }
+          },
+          "output": {
+            "force-delay": {
+              "min": 0,
+              "max": 0
+            },
+            "misc": {
+              "at-sender": false,
+              "quote-origin": false
+            }
+          },
+          "trigger": {
+            "group-respond-rules": {
+              "prefix": [
+                "ai "
+              ]
+            }
+          }
+        }
+      }
+    },
+    {
+      "id": "chat_bot",
+      "kind": "bot",
+      "body": {
+        "name": "regression group chat",
+        "description": "Synthetic signed group ingress",
+        "adapter": "http_bot",
+        "enable": true,
+        "adapter_config": {
+          "inbound_secret": "${group_secret}",
+          "signature_required": true,
+          "callback_url": "http://127.0.0.1:9/unused-sync-only",
+          "default_session_type": "group",
+          "callback_timeout": 20,
+          "callback_max_retries": 0
+        },
+        "event_bindings": [
+          {
+            "event_pattern": "message.received",
+            "target_type": "pipeline",
+            "target_uuid": "${chat_pipeline}",
+            "enabled": true,
+            "filters": []
+          }
+        ],
+        "plugin_processors": []
+      }
+    },
+    {
+      "id": "box_bot",
+      "kind": "bot",
+      "body": {
+        "name": "regression group box",
+        "description": "Synthetic signed group ingress",
+        "adapter": "http_bot",
+        "enable": true,
+        "adapter_config": {
+          "inbound_secret": "${group_secret}",
+          "signature_required": true,
+          "callback_url": "http://127.0.0.1:9/unused-sync-only",
+          "default_session_type": "group",
+          "callback_timeout": 20,
+          "callback_max_retries": 0
+        },
+        "event_bindings": [
+          {
+            "event_pattern": "message.received",
+            "target_type": "pipeline",
+            "target_uuid": "${box_pipeline}",
+            "enabled": true,
+            "filters": []
+          }
+        ],
+        "plugin_processors": []
+      }
+    },
+    {
+      "id": "rag_bot",
+      "kind": "bot",
+      "body": {
+        "name": "regression group rag",
+        "description": "Synthetic signed group ingress",
+        "adapter": "http_bot",
+        "enable": true,
+        "adapter_config": {
+          "inbound_secret": "${group_secret}",
+          "signature_required": true,
+          "callback_url": "http://127.0.0.1:9/unused-sync-only",
+          "default_session_type": "group",
+          "callback_timeout": 20,
+          "callback_max_retries": 0
+        },
+        "event_bindings": [
+          {
+            "event_pattern": "message.received",
+            "target_type": "pipeline",
+            "target_uuid": "${rag_pipeline}",
+            "enabled": true,
+            "filters": []
+          }
+        ],
+        "plugin_processors": []
+      }
+    }
+  ],
+  "cases": [
+    {
+      "name": "chat",
+      "weight": 2,
+      "http_group": {
+        "url": "${core_base_url}/bots/${chat_bot}/sync",
+        "secret_env": "LANGBOT_QA_GROUP_SECRET",
+        "message": "ai What is the documented parser-rag sentinel?"
+      },
+      "expect": "LANGBOT_CHAT_OK"
+    },
+    {
+      "name": "box",
+      "weight": 1,
+      "http_group": {
+        "url": "${core_base_url}/bots/${box_bot}/sync",
+        "secret_env": "LANGBOT_QA_GROUP_SECRET",
+        "message": "ai What is the documented parser-rag sentinel?"
+      },
+      "expect": "LANGBOT_BOX_OK"
+    },
+    {
+      "name": "rag_chat",
+      "weight": 1,
+      "http_group": {
+        "url": "${core_base_url}/bots/${rag_bot}/sync",
+        "secret_env": "LANGBOT_QA_GROUP_SECRET",
+        "message": "ai What is the documented parser-rag sentinel?"
+      },
+      "expect": "LANGBOT_RAG_OK"
+    }
+  ],
+  "phases": [
+    {
+      "name": "normal",
+      "duration_seconds": 600,
+      "rps": 2,
+      "workers": 4,
+      "cooldown_seconds": 30,
+      "groups": 4,
+      "members_per_group": 3
+    }
+  ],
+  "observe": {
+    "health": {
+      "core": "${core_health}",
+      "mock": "http://127.0.0.1:18080/healthz",
+      "box": "http://127.0.0.1:5410/readyz"
+    },
+    "pids": {},
+    "pid_files": {},
+    "containers": []
+  },
+  "cleanup": "on-success",
+  "thresholds": {
+    "min_rps_ratio": 0,
+    "max_p95_seconds": {
+      "chat": 5,
+      "box": 10,
+      "rag_chat": 6
+    }
+  }
+}

tools/regression_profiles/job.json

@@ -1,9 +1,14 @@
 {
   "version": 1,
   "lbctl": "lbctl",
-  "suite": "normal.json",
-  "deployment": {"version": "v4.11.0", "port": 15351},
-  "vars": {"plugin_dir": "/path/to/plugin-zips"},
+  "suite": "groups.json",
+  "deployment": {
+    "version": "v4.11.0",
+    "port": 15351
+  },
+  "vars": {
+    "plugin_dir": "/path/to/plugin-zips"
+  },
   "mock_image": "python:3.12-alpine",
   "cleanup": "on-success",
   "purge_data": true

tools/regression_suite.py

@@ -8,11 +8,17 @@ import math
 import os
 from pathlib import Path
 import re
+import secrets
 import subprocess
 import sys
+import urllib.error
+import urllib.request
+from uuid import uuid4
+
+from regression_traffic import group_request
 
 
-KINDS = {'provider', 'model', 'pipeline', 'knowledge-base'}
+KINDS = {'provider', 'model', 'pipeline', 'knowledge-base', 'bot'}
 REFERENCE = re.compile(r'\$\{([a-zA-Z][a-zA-Z0-9_-]*)\}')
 
 
@@ -35,6 +41,21 @@ def merge(base, patch):
     return result
 
 
+def contains_fields(actual, expected):
+    """Check submitted fields while allowing server-generated defaults."""
+    if isinstance(expected, dict):
+        return isinstance(actual, dict) and all(
+            key in actual and contains_fields(actual[key], value) for key, value in expected.items()
+        )
+    if isinstance(expected, list):
+        return (
+            isinstance(actual, list)
+            and len(actual) == len(expected)
+            and all(contains_fields(left, right) for left, right in zip(actual, expected))
+        )
+    return actual == expected
+
+
 def number(value, label, *, minimum=0, integer=False):
     if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
         raise ValueError(f'{label} must be a finite number')
@@ -61,7 +82,7 @@ def validate(profile):
         raise ValueError('cli accepts config and executable')
     if not isinstance(cli.get('config'), str) or not cli['config']:
         raise ValueError('cli.config must name a Workspace-bound lbctl config')
-    variables = {'profile_dir': '.', 'run_dir': '.', 'workspace_uuid': 'workspace'}
+    variables = {'profile_dir': '.', 'run_dir': '.', 'workspace_uuid': 'workspace', 'group_secret': 'private'}
     supplied = profile.get('vars', {})
     if not isinstance(supplied, dict) or variables.keys() & supplied.keys():
         raise ValueError('vars must be an object without reserved names')
@@ -84,6 +105,16 @@ def validate(profile):
             raise ValueError(f'{name}: resource needs kind and body')
         if 'uuid' in item['body']:
             raise ValueError(f'{name}: resources must create new objects; body.uuid is forbidden')
+        if item['kind'] == 'bot' and (
+            item['body'].get('adapter') != 'http_bot'
+            or not isinstance(item['body'].get('adapter_config'), dict)
+            or item['body'].get('adapter_config', {}).get('inbound_secret') != '${group_secret}'
+            or not any(
+                isinstance(case.get('http_group'), dict) and '${' + name + '}' in case['http_group'].get('url', '')
+                for case in profile.get('cases', [])
+            )
+        ):
+            raise ValueError('Bot resources require http_bot, ${group_secret}, and a matching http_group case')
         if item['kind'] == 'model' and item.get('type') not in {'llm', 'embedding', 'rerank'}:
             raise ValueError(f'{name}: model type must be llm/embedding/rerank')
         resolve(item['body'], variables)  # References must point backwards; never guess missing UUIDs.
@@ -100,12 +131,26 @@ def validate(profile):
         raise ValueError('cases must be a non-empty array')
     names = set()
     for case in cases:
-        if case.keys() - {'name', 'weight', 'args', 'expect', 'body'}:
+        if case.keys() - {'name', 'weight', 'args', 'expect', 'body', 'http_group'}:
             raise ValueError('unknown case field')
         if not isinstance(case.get('name'), str) or not case['name'] or case['name'] in names:
             raise ValueError('case names must be non-empty and unique')
         names.add(case['name'])
-        if (
+        if 'http_group' in case:
+            settings = case['http_group']
+            if (
+                'args' in case
+                or 'body' in case
+                or not isinstance(settings, dict)
+                or set(settings) != {'url', 'secret_env', 'message'}
+                or any(not isinstance(value, str) or not value for value in settings.values())
+                or not resolve(settings['url'], variables).startswith(('http://', 'https://'))
+                or settings['secret_env'] != 'LANGBOT_QA_GROUP_SECRET'
+            ):
+                raise ValueError(
+                    'http_group needs url, message and secret_env=LANGBOT_QA_GROUP_SECRET; no CLI args/body'
+                )
+        elif (
             not isinstance(case.get('args'), list)
             or not case['args']
             or not all(isinstance(x, str) for x in case['args'])
@@ -128,7 +173,15 @@ def validate(profile):
         if not re.fullmatch(r'[a-zA-Z][a-zA-Z0-9_-]*', name) or name in names:
             raise ValueError('phase names must be unique simple names')
         names.add(name)
-        if phase.keys() - {'name', 'duration_seconds', 'rps', 'workers', 'cooldown_seconds'}:
+        if phase.keys() - {
+            'name',
+            'duration_seconds',
+            'rps',
+            'workers',
+            'cooldown_seconds',
+            'groups',
+            'members_per_group',
+        }:
             raise ValueError(f'{name}: unknown phase field')
         number(phase.get('duration_seconds'), 'duration_seconds', minimum=0.001)
         number(phase.get('rps'), 'rps', minimum=0.001)
@@ -136,6 +189,17 @@ def validate(profile):
         number(phase.get('cooldown_seconds', 0), 'cooldown_seconds')
         if phase['workers'] > 32:
             raise ValueError('workers must be <= 32')
+        number(phase.get('groups', 0), 'groups', integer=True)
+        number(phase.get('members_per_group', 3), 'members_per_group', minimum=1, integer=True)
+        if phase.get('groups', 0) > 4096 or phase.get('members_per_group', 3) > 1000:
+            raise ValueError('groups must be <= 4096 and members_per_group <= 1000')
+        group_cases = ['http_group' in case for case in cases]
+        if any(group_cases) and (
+            not all(group_cases) or phase.get('groups', 0) < sum(case.get('weight', 1) for case in cases)
+        ):
+            raise ValueError('group phases need only http_group cases and enough groups for all weighted cases')
+        if not any(group_cases) and phase.get('groups', 0):
+            raise ValueError('groups require http_group cases')
     if profile.get('cleanup', 'on-success') not in {'always', 'on-success', 'never'}:
         raise ValueError('cleanup must be always/on-success/never')
     thresholds = profile.get('thresholds', {})
@@ -151,8 +215,18 @@ def validate(profile):
             raise ValueError(f'unknown case in max_p95_seconds: {name}')
         number(value, 'max_p95_seconds', minimum=0.001)
     observed = resolve(profile.get('observe', {}), variables)
-    if not isinstance(observed, dict) or observed.keys() - {'health', 'pids', 'pid_files', 'containers'}:
-        raise ValueError('observe accepts health, pids, pid_files, containers')
+    if not isinstance(observed, dict) or observed.keys() - {
+        'health',
+        'pids',
+        'pid_files',
+        'containers',
+        'box_data_root',
+    }:
+        raise ValueError('observe accepts health, pids, pid_files, containers, box_data_root')
+    if 'box_data_root' in observed and (
+        not isinstance(observed['box_data_root'], str) or not observed['box_data_root']
+    ):
+        raise ValueError('observe.box_data_root must name the deployment Box data directory')
     if not isinstance(observed.get('health'), dict) or not observed['health']:
         raise ValueError('observe.health must contain at least one health endpoint')
     for name, url in observed['health'].items():
@@ -186,8 +260,14 @@ class Suite:
             'json',
         ]
         self.owned = []
-        self.env = env
-        self.secrets = [value for key, value in (env or os.environ).items() if 'API_KEY' in key and value]
+        self.connection = {}
+        self.env = dict(os.environ if env is None else env)
+        if any('http_group' in case for case in profile['cases']):
+            self.variables['group_secret'] = secrets.token_urlsafe(32)
+            self.env['LANGBOT_QA_GROUP_SECRET'] = self.variables['group_secret']
+        self.secrets = [
+            value for key, value in self.env.items() if ('API_KEY' in key or key == 'LANGBOT_QA_GROUP_SECRET') and value
+        ]
 
     def path(self, value):
         path = Path(resolve(value, self.variables)).expanduser()
@@ -195,11 +275,15 @@ class Suite:
 
     def write(self, name, value):
         path = self.output / name
+        if name == 'request.json':
+            private = self.output / '.private'
+            private.mkdir(exist_ok=True, mode=0o700)
+            path = private / name
         path.write_text(json.dumps(value, ensure_ascii=False, indent=2))
         path.chmod(0o600)
         return str(path)
 
-    def call(self, args, body=None):
+    def call(self, args, body=None, *, defer_bot_readback=False):
         command = [*self.cli, *args]
         if body is not None:
             command += ['--file', self.write('request.json', body)]
@@ -208,11 +292,50 @@ class Suite:
         except subprocess.TimeoutExpired as exc:
             raise RuntimeError('lbctl timed out; outcome unknown, no automatic retry') from exc
         if result.returncode:
+            if defer_bot_readback and args == ['bot', 'create']:
+                response = json.loads(result.stdout)
+                data = response.get('data', {})
+                if (
+                    response.get('error', {}).get('type') == 'verification_failed'
+                    and data.get('operation') == 'bot.create'
+                    and data.get('uuid')
+                    and data.get('bot', {}).get('uuid') == data['uuid']
+                ):
+                    return data
             error = result.stdout + result.stderr
             for secret in self.secrets:
                 error = error.replace(secret, '[REDACTED]')
             raise RuntimeError(error[:3000])
-        return json.loads(result.stdout)['data']
+        response = json.loads(result.stdout)
+        if args == ['whoami']:
+            self.connection = response.get('meta') or {}
+        return response['data']
+
+    def read_bot(self, bot_uuid):
+        """Read full Bot fields using the endpoint and environment credential confirmed by lbctl."""
+        source = self.connection.get('credential_source', '')
+        key = self.env.get(source.removeprefix('env:')) if source.startswith('env:') else None
+        if not key or not self.connection.get('endpoint'):
+            raise RuntimeError('Group Bot readback requires environment-backed lbctl credentials')
+        self.secrets.append(key)
+        request = urllib.request.Request(
+            self.connection['endpoint'].rstrip('/') + '/api/v1/platform/bots/' + bot_uuid,
+            headers={'Authorization': 'Bearer ' + key, 'X-Workspace-Id': self.variables['workspace_uuid']},
+        )
+        opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+        try:
+            with opener.open(request, timeout=30) as response:
+                payload = json.load(response)
+        except (urllib.error.URLError, OSError) as exc:
+            raise RuntimeError('Bot readback request failed; the creation was not retried') from exc
+        if (
+            not isinstance(payload, dict)
+            or payload.get('code') != 0
+            or not isinstance(payload.get('data'), dict)
+            or not isinstance(payload['data'].get('bot'), dict)
+        ):
+            raise RuntimeError('Bot readback did not return a successful Bot resource')
+        return payload['data']['bot']
 
     def journal(self):
         self.write('owned.json', {'workspace_uuid': self.variables['workspace_uuid'], 'resources': self.owned})
@@ -230,12 +353,19 @@ class Suite:
             args = [kind, 'apply' if kind == 'pipeline' else 'create']
             if kind == 'model':
                 args += ['--type', item['type']]
-            created = self.call(args, body)
+            created = self.call(args, body, defer_bot_readback=True) if kind == 'bot' else self.call(args, body)
             created = created.get('pipeline', created) if kind == 'pipeline' else created
             uuid = created['uuid']
             self.variables[item['id']] = uuid
             self.owned.append({'kind': kind, 'uuid': uuid, **({'type': item['type']} if kind == 'model' else {})})
             self.journal()  # Record ownership before any later configuration or ingestion can fail.
+            if kind == 'bot':
+                # lbctl 0.3's Bot projection omits event bindings; verify the existing Core API instead.
+                actual = self.read_bot(uuid)
+                expected = copy.deepcopy(body)
+                expected['adapter_config'].pop('inbound_secret')
+                if not contains_fields(actual, expected):
+                    raise RuntimeError(f'Bot readback differs from submitted non-secret fields: {uuid}')
             if config is not None:
                 current = self.call(['pipeline', 'get', uuid])
                 current = current.get('pipeline', current)
@@ -263,8 +393,20 @@ class Suite:
                 )
         self.write(
             'resources.json',
-            {key: value for key, value in self.variables.items() if key not in self.profile.get('vars', {})},
+            {
+                key: value
+                for key, value in self.variables.items()
+                if key not in self.profile.get('vars', {}) and key != 'group_secret'
+            },
         )
+        warmups = []
+        for case in self.profile['cases']:
+            if 'http_group' in case:
+                case = resolve(case, self.variables)
+                group = f'qa-warmup-{uuid4().hex[:12]}'
+                response = group_request(case, group, 'qa-member-000', 1, uuid4().hex, 'none', self.env)
+                warmups.append({'case': case['name'], 'output': response})
+                self.write('warmup.json', warmups)
 
     def cleanup(self):
         saved = json.loads((self.output / 'owned.json').read_text())
@@ -291,7 +433,10 @@ class Suite:
             case = resolve(original, self.variables)
             if 'body' in case:
                 case['args'] += ['--file', self.write(f'case-{len(cases)}.json', case.pop('body'))]
-            cases.extend([{key: case[key] for key in ('name', 'args', 'expect')}] * case.get('weight', 1))
+            cases.extend(
+                [{key: case[key] for key in ('name', 'args', 'expect', 'http_group') if key in case}]
+                * case.get('weight', 1)
+            )
         cases_path = self.write('cases.json', cases)
         observed = resolve(self.profile.get('observe', {}), self.variables)
         observe_args = []
@@ -303,6 +448,8 @@ class Suite:
             observe_args += ['--health', f'{name}={url}']
         for container in observed.get('containers', []):
             observe_args += ['--container', container]
+        if observed.get('box_data_root'):
+            observe_args += ['--box-data-root', self.path(observed['box_data_root'])]
         results = []
         errors = []
         for phase in self.profile['phases']:
@@ -327,6 +474,12 @@ class Suite:
                 '--cooldown',
                 str(phase.get('cooldown_seconds', 0)),
                 '--shared-case-sequence',
+                '--groups',
+                str(phase.get('groups', 0)),
+                '--members-per-group',
+                str(phase.get('members_per_group', 3)),
+                '--group-state',
+                str(self.output / 'group-state.json'),
                 *observe_args,
             ]
             with (self.output / f'{phase["name"]}.log').open('w') as log:
@@ -355,11 +508,19 @@ def main():
     parser.add_argument('--duration', type=float, help='Override duration_seconds in every phase')
     parser.add_argument('--rps', type=float, help='Override RPS in every phase')
     parser.add_argument('--workers', type=int, help='Override workers in every phase')
+    parser.add_argument('--groups', type=int, help='Override group count in every group phase')
+    parser.add_argument('--members-per-group', type=int)
     args = parser.parse_args()
     try:
         profile = json.loads(args.profile.read_text())
         for phase in profile.get('phases', []):
-            for flag, field in [('duration', 'duration_seconds'), ('rps', 'rps'), ('workers', 'workers')]:
+            for flag, field in [
+                ('duration', 'duration_seconds'),
+                ('rps', 'rps'),
+                ('workers', 'workers'),
+                ('groups', 'groups'),
+                ('members_per_group', 'members_per_group'),
+            ]:
                 if getattr(args, flag) is not None:
                     phase[field] = getattr(args, flag)
         validate(profile)

tools/regression_traffic.py

@@ -1,10 +1,12 @@
 #!/usr/bin/env python3
-"""Drive normal lbctl traffic and retain business checks, health and Linux resources."""
+"""Drive independent CLI requests or persistent group conversations with Linux resource sampling."""
 
 import argparse
 from collections import Counter
 from concurrent.futures import ThreadPoolExecutor
 import json
+import hashlib
+import hmac
 import math
 import os
 from pathlib import Path
@@ -12,6 +14,9 @@ import subprocess
 import sys
 import threading
 import time
+import urllib.error
+import urllib.request
+import uuid
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
 from cloud_runtime_soak import read_endpoint_snapshot, read_process_snapshot
@@ -21,19 +26,97 @@ def percentile(values, fraction):
     return sorted(values)[max(0, math.ceil(len(values) * fraction) - 1)] if values else None
 
 
+def owned_box_containers(data_root, checked=None):
+    """Discover sandboxes mounted below one run's data root, including newly created groups."""
+    if checked is None:
+
+        def checked(command):
+            return subprocess.check_output(command, text=True, timeout=10)
+
+    candidates = checked(['docker', 'ps', '-a', '--filter', 'label=langbot.box=true', '--format', '{{.ID}}']).split()
+    if not candidates:
+        return []
+    root = Path(data_root).resolve()
+    descriptions = checked(['docker', 'inspect', '--format', '{{json .}}', *candidates])
+    owned = []
+    for line in descriptions.splitlines():
+        if not line:
+            continue
+        entry = json.loads(line)
+        if any(
+            Path(mount['Source']).resolve().is_relative_to(root) for mount in entry['Mounts'] if mount.get('Source')
+        ):
+            owned.append(entry['Id'])
+    return owned
+
+
+def group_request(case, group_id, member_id, turn, token, previous, env=None):
+    """Send one signed GroupMessage through the existing HTTP Bot adapter."""
+    settings = case['http_group']
+    secret = (os.environ if env is None else env)[settings['secret_env']]
+    marker = f'LB_GROUP={group_id} LB_TURN={turn} LB_TOKEN={token} LB_PREVIOUS={previous}'
+    body = json.dumps(
+        {
+            'session_id': group_id,
+            'session_type': 'group',
+            'sender': {'id': member_id, 'name': member_id, 'group_name': group_id},
+            'message': [{'type': 'Plain', 'text': settings['message'] + ' ' + marker}],
+        }
+    ).encode()
+    stamp = str(int(time.time()))
+    signature = hmac.new(secret.encode(), stamp.encode() + b'.' + body, hashlib.sha256).hexdigest()
+    request = urllib.request.Request(
+        settings['url'],
+        data=body,
+        headers={
+            'Content-Type': 'application/json',
+            'X-LB-Timestamp': stamp,
+            'X-LB-Signature': 'sha256=' + signature,
+            'X-LB-Idempotency-Key': token,
+        },
+    )
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    with opener.open(request, timeout=90) as response:
+        payload = json.load(response)
+        if response.status != 200 or not isinstance(payload, dict) or payload.get('code') != 0:
+            raise ValueError('group ingress did not return a completed response')
+    data = payload.get('data', {})
+    if not isinstance(data, dict):
+        raise ValueError('group ingress response data must be an object')
+    parts = data.get('message')
+    if not isinstance(parts, list) or any(not isinstance(part, dict) for part in parts):
+        raise ValueError('group reply must contain message components')
+    text = ''.join(str(part.get('text', '')) for part in parts if part.get('type') == 'Plain')
+    if data.get('session_id') != group_id or not data.get('reply_to'):
+        raise ValueError('group reply session or message correlation is missing')
+    if case['expect'] not in text or f'GROUP_ACK={token}' not in text:
+        raise ValueError('group reply failed its business or conversation-history oracle')
+    return payload
+
+
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--lbctl', required=True)
     parser.add_argument('--config', required=True)
-    parser.add_argument('--cases', type=Path, required=True, help='JSON array of {name, args, expect}')
+    parser.add_argument(
+        '--cases', type=Path, required=True, help='JSON cases with name, expect, and args or http_group'
+    )
     parser.add_argument('--output', type=Path, required=True)
     parser.add_argument('--duration', type=float, default=600)
     parser.add_argument('--rps', type=float, default=2)
     parser.add_argument('--workers', type=int, default=4)
+    parser.add_argument(
+        '--groups', type=int, default=0, help='Stable group sessions; zero uses independent CLI requests'
+    )
+    parser.add_argument('--members-per-group', type=int, default=3)
+    parser.add_argument(
+        '--group-state', type=Path, help='Carry group histories across load phases; refuses interrupted runs'
+    )
     parser.add_argument('--cooldown', type=float, default=30)
     parser.add_argument('--pid', action='append', default=[], metavar='NAME=PID')
     parser.add_argument('--health', action='append', default=[], metavar='NAME=URL')
     parser.add_argument('--container', action='append', default=[])
+    parser.add_argument('--box-data-root', type=Path, help='Dynamically sample Box containers mounted under this root')
     parser.add_argument(
         '--shared-case-sequence', action='store_true', help='Share the weighted case sequence across workers'
     )
@@ -46,6 +129,8 @@ def main():
         or args.cooldown < 0
     ):
         parser.error('duration/rps must be positive, workers 1..32, cooldown nonnegative')
+    if not 0 <= args.groups <= 4096 or not 1 <= args.members_per_group <= 1000:
+        parser.error('groups must be 0..4096 (0 selects CLI mode), members-per-group 1..1000')
     cases = json.loads(args.cases.read_text())
     if (
         not isinstance(cases, list)
@@ -53,26 +138,77 @@ def main():
         or any(
             not isinstance(case, dict)
             or not isinstance(case.get('name'), str)
-            or not isinstance(case.get('args'), list)
-            or not all(isinstance(item, str) for item in case['args'])
+            or (
+                not args.groups
+                and (not isinstance(case.get('args'), list) or not all(isinstance(item, str) for item in case['args']))
+            )
             or not isinstance(case.get('expect'), str)
             or not case['expect']
             for case in cases
         )
     ):
         parser.error('cases must contain name, string args, and expected response text')
+    if args.groups:
+        if args.groups < len(cases):
+            parser.error('groups must cover the complete weighted case sequence')
+        for case in cases:
+            settings = case.get('http_group', {})
+            if (
+                not isinstance(settings, dict)
+                or set(settings) != {'url', 'secret_env', 'message'}
+                or not all(isinstance(value, str) and value for value in settings.values())
+                or not settings['url'].startswith(('http://', 'https://'))
+                or not os.environ.get(settings['secret_env'])
+            ):
+                parser.error('group cases need HTTP(S) url, message and a populated secret_env')
     args.output.mkdir(parents=True, exist_ok=False)
     pids = {name: int(pid) for name, pid in (value.split('=', 1) for value in args.pid)}
     endpoints = dict(value.split('=', 1) for value in args.health)
     rows = []
     sampling_errors = []
     lock = threading.Lock()
+    group_condition = threading.Condition(lock)
     stop = threading.Event()
     phase = ['load']
     next_case = 0
     started = time.monotonic()
     deadline = started + args.duration
     redactions = [value for name, value in os.environ.items() if 'API_KEY' in name and value]
+    group_state = None
+    groups = []
+    if args.groups:
+        redactions += [os.environ[case['http_group']['secret_env']] for case in cases]
+        state_path = args.group_state or args.output / 'group-state.json'
+        fingerprint = [(case['name'], case['http_group']) for case in cases]
+        group_state = (
+            json.loads(state_path.read_text())
+            if state_path.exists()
+            else {
+                'run_id': uuid.uuid4().hex[:12],
+                'cases': fingerprint,
+                'groups': [],
+                'in_progress': False,
+            }
+        )
+        if group_state.get('in_progress') or json.dumps(group_state['cases']) != json.dumps(fingerprint):
+            parser.error('group state is interrupted or belongs to different cases; inspect it before reuse')
+        for index in range(len(group_state['groups']), args.groups):
+            group_state['groups'].append(
+                {
+                    'id': f'qa-{group_state["run_id"]}-g{index:04d}',
+                    'case': index % len(cases),
+                    'turn': 1,
+                    'previous': 'none',
+                    'failed': False,
+                }
+            )
+        groups = [
+            dict(item, due=started + index / args.rps, busy=False)
+            for index, item in enumerate(group_state['groups'][: args.groups])
+        ]
+        group_state['in_progress'] = True
+        state_path.write_text(json.dumps(group_state, indent=2))
+        state_path.chmod(0o600)
 
     def redact(text):
         for secret in redactions:
@@ -99,9 +235,12 @@ def main():
                         row['health'][name] = read_endpoint_snapshot(url, timeout_seconds=3)
                     except RuntimeError as exc:
                         row['health'][name] = {'error': str(exc)}
-                if args.container:
+                containers = set(args.container)
+                if args.box_data_root:
+                    containers.update(owned_box_containers(args.box_data_root))
+                if containers:
                     result = subprocess.run(
-                        ['docker', 'stats', '--no-stream', '--format', '{{json .}}', *args.container],
+                        ['docker', 'stats', '--no-stream', '--format', '{{json .}}', *sorted(containers)],
                         capture_output=True,
                         text=True,
                         timeout=10,
@@ -163,6 +302,60 @@ def main():
             iteration += 1
             due = max(due + args.workers / args.rps, time.monotonic())
 
+    def group_worker(_index):
+        while not stop.is_set():
+            with group_condition:
+                while True:
+                    now = time.monotonic()
+                    if stop.is_set() or now >= deadline:
+                        return
+                    available = [group for group in groups if not group['busy'] and not group['failed']]
+                    if not available:
+                        if not any(group['busy'] for group in groups):
+                            return
+                        group_condition.wait(0.1)
+                        continue
+                    group = min(available, key=lambda item: item['due'])
+                    if group['due'] > now:
+                        group_condition.wait(min(1, group['due'] - now))
+                        continue
+                    group['busy'] = True
+                    break
+            case = cases[group['case']]
+            before = time.monotonic()
+            token = uuid.uuid4().hex
+            member = f'qa-member-{(group["turn"] - 1) % args.members_per_group:03d}'
+            row = {
+                'case': case['name'],
+                'group_id': group['id'],
+                'member_id': member,
+                'turn': group['turn'],
+                'time': time.time(),
+                'elapsed': before - started,
+                'status': 'failed',
+            }
+            try:
+                row['output'] = group_request(case, group['id'], member, group['turn'], token, group['previous'])
+                row['status'] = 'ok'
+            except urllib.error.HTTPError as exc:
+                row.update(http_status=exc.code, error=redact(exc.read(2000).decode(errors='replace')))
+                if exc.code >= 500:
+                    row['status'] = 'unknown'
+            except (urllib.error.URLError, TimeoutError, OSError) as exc:
+                row.update(status='unknown', error=redact(str(exc)) + '; not retried')
+            except (ValueError, KeyError) as exc:
+                row['error'] = redact(str(exc))
+            row['seconds'] = time.monotonic() - before
+            with group_condition:
+                rows.append(row)
+                with (args.output / 'requests.jsonl').open('a') as stream:
+                    stream.write(json.dumps(row, ensure_ascii=False) + '\n')
+                group.update(busy=False, failed=row['status'] != 'ok')
+                if row['status'] == 'ok':
+                    group.update(turn=group['turn'] + 1, previous=token)
+                group['due'] = max(group['due'] + args.groups / args.rps, time.monotonic())
+                group_condition.notify_all()
+
     def sampling_guard():
         try:
             sample()
@@ -174,7 +367,7 @@ def main():
     sampler.start()
     try:
         with ThreadPoolExecutor(max_workers=args.workers) as pool:
-            list(pool.map(worker, range(args.workers)))
+            list(pool.map(group_worker if args.groups else worker, range(args.workers)))
         stop.wait(max(0, deadline - time.monotonic()))
         load_elapsed = time.monotonic() - started
         phase[0] = 'cooldown'
@@ -200,6 +393,36 @@ def main():
         for sample in samples
     )
     summary['sampling_errors'] = sampling_errors
+    if args.groups:
+        for index, group in enumerate(groups):
+            group_state['groups'][index] = {key: value for key, value in group.items() if key not in {'due', 'busy'}}
+        group_state['in_progress'] = False
+        temporary = state_path.with_suffix('.tmp')
+        temporary.write_text(json.dumps(group_state, indent=2))
+        temporary.chmod(0o600)
+        temporary.replace(state_path)
+        summary['groups'] = {
+            'configured': args.groups,
+            'members_per_group': args.members_per_group,
+            'target_rps_per_group': args.rps / args.groups,
+            'sessions': {
+                group['id']: {
+                    'case': cases[group['case']]['name'],
+                    'successful_turns_total': group['turn'] - 1,
+                    'failed': group['failed'],
+                    'requests_this_phase': sum(row.get('group_id') == group['id'] for row in rows),
+                    'members_this_phase': sorted(
+                        {row['member_id'] for row in rows if row.get('group_id') == group['id']}
+                    ),
+                }
+                for group in groups
+            },
+        }
+        summary['group_failures'] = [
+            group_id
+            for group_id, result in summary['groups']['sessions'].items()
+            if result['failed'] or result['successful_turns_total'] < 2 or not result['requests_this_phase']
+        ]
     for name in dict.fromkeys(case['name'] for case in cases):
         selected = [row for row in rows if row['case'] == name]
         times = [row['seconds'] for row in selected if row['status'] == 'ok']
@@ -212,7 +435,13 @@ def main():
         }
     (args.output / 'summary.json').write_text(json.dumps(summary, indent=2))
     print(json.dumps(summary, indent=2), flush=True)
-    if not rows or any(row['status'] != 'ok' for row in rows) or summary['health_failures'] or sampling_errors:
+    if (
+        not rows
+        or any(row['status'] != 'ok' for row in rows)
+        or summary['health_failures']
+        or sampling_errors
+        or summary.get('group_failures')
+    ):
         raise SystemExit(1)
 
 

tools/test_mock_llm.py

@@ -66,6 +66,19 @@ class MockLLMTest(unittest.TestCase):
         self.assertEqual(content, 'LANGBOT_CHAT_OK')
         self.assertEqual(chunks[-1]['usage']['total_tokens'], 15)
 
+    def test_group_history_requires_previous_user_reply_and_isolation(self):
+        first = {'role': 'user', 'content': 'LB_GROUP=group-a LB_TURN=1 LB_TOKEN=aaa LB_PREVIOUS=none'}
+        second = {'role': 'user', 'content': 'LB_GROUP=group-a LB_TURN=2 LB_TOKEN=bbb LB_PREVIOUS=aaa'}
+        reply = json.loads(self.post(messages=[first]))['choices'][0]['message']
+        self.assertIn('GROUP_ACK=aaa', reply['content'])
+        result = json.loads(self.post(messages=[first, reply, second]))['choices'][0]['message']
+        self.assertIn('GROUP_ACK=bbb', result['content'])
+        other = {'role': 'user', 'content': 'LB_GROUP=group-b LB_TURN=1 LB_TOKEN=ccc LB_PREVIOUS=none'}
+        for messages in ([second], [first, second], [other, reply, second]):
+            with self.subTest(messages=messages), self.assertRaises(HTTPError) as failure:
+                self.post(messages=messages)
+            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}

tools/test_regression_groups.py

@@ -0,0 +1,240 @@
+"""Exercise signed group traffic, member rotation, phase continuity and fail-closed history."""
+
+import hashlib
+import hmac
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+
+from regression_suite import Suite, validate
+
+
+class GroupTests(unittest.TestCase):
+    def test_bot_readback_uses_the_cli_confirmed_endpoint_and_workspace(self):
+        received = []
+
+        class Readback(BaseHTTPRequestHandler):
+            def do_GET(self):
+                received.append((self.path, self.headers.get('Authorization'), self.headers.get('X-Workspace-Id')))
+                body = json.dumps({'code': 0, 'data': {'bot': {'uuid': 'owned-bot', 'event_bindings': []}}}).encode()
+                self.send_response(200)
+                self.end_headers()
+                self.wfile.write(body)
+
+            def log_message(self, *_args):
+                pass
+
+        server = ThreadingHTTPServer(('127.0.0.1', 0), Readback)
+        thread = threading.Thread(target=server.serve_forever)
+        thread.start()
+        try:
+            with tempfile.TemporaryDirectory() as folder:
+                current = json.loads((Path(__file__).with_name('regression_profiles') / 'groups.json').read_text())
+                suite = Suite(current, Path(folder) / 'profile.json', Path(folder), env={'TEST_API_KEY': 'test-key'})
+                suite.variables['workspace_uuid'] = 'bound-workspace'
+                suite.connection = {
+                    'endpoint': f'http://127.0.0.1:{server.server_port}',
+                    'credential_source': 'env:TEST_API_KEY',
+                }
+                self.assertEqual(suite.read_bot('owned-bot')['uuid'], 'owned-bot')
+                self.assertEqual(received, [('/api/v1/platform/bots/owned-bot', 'Bearer test-key', 'bound-workspace')])
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join()
+
+    def test_groups_profile_and_invalid_modes(self):
+        path = Path(__file__).with_name('regression_profiles') / 'groups.json'
+        profile = json.loads(path.read_text())
+        validate(profile)
+        for field, value in [('groups', 3), ('members_per_group', 0), ('groups', 4097)]:
+            current = json.loads(path.read_text())
+            current['phases'][0][field] = value
+            with self.subTest(field=field), self.assertRaises(ValueError):
+                validate(current)
+
+    def test_signed_groups_continue_across_phases_and_stop_after_bad_reply(self):
+        secret = 'test-group-signing-key'
+        histories = {}
+        failures = []
+        active = set()
+        lock = threading.Lock()
+        bad_reply = False
+
+        class Ingress(BaseHTTPRequestHandler):
+            def do_GET(self):
+                self.respond({'code': 0})
+
+            def respond(self, payload, status=200):
+                body = json.dumps(payload).encode()
+                self.send_response(status)
+                self.send_header('Content-Length', str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+
+            def do_POST(self):
+                raw = self.rfile.read(int(self.headers['Content-Length']))
+                stamp = self.headers['X-LB-Timestamp']
+                expected = (
+                    'sha256=' + hmac.new(secret.encode(), stamp.encode() + b'.' + raw, hashlib.sha256).hexdigest()
+                )
+                payload = json.loads(raw)
+                group = payload['session_id']
+                fields = dict(re.findall(r'LB_(GROUP|TURN|TOKEN|PREVIOUS)=(\S+)', payload['message'][0]['text']))
+                with lock:
+                    prior = histories.get(group, [])
+                    if (
+                        self.headers['X-LB-Signature'] != expected
+                        or group in active
+                        or payload['session_type'] != 'group'
+                        or fields['GROUP'] != group
+                        or int(fields['TURN']) != len(prior) + 1
+                        or fields['PREVIOUS'] != (prior[-1]['token'] if prior else 'none')
+                    ):
+                        failures.append(payload)
+                    active.add(group)
+                time.sleep(0.01)
+                with lock:
+                    histories.setdefault(group, []).append(
+                        {'token': fields['TOKEN'], 'member': payload['sender']['id']}
+                    )
+                    active.remove(group)
+                ack = 'missing' if bad_reply and group.endswith('g0000') else fields['TOKEN']
+                self.respond(
+                    {
+                        'code': 0,
+                        'data': {
+                            'session_id': group,
+                            'reply_to': 'incoming-' + fields['TOKEN'],
+                            'message': [
+                                {'type': 'Plain', 'text': 'TEST_OK GROUP_'},
+                                {'type': 'Plain', 'text': 'ACK=' + ack},
+                            ],
+                        },
+                    },
+                    status=500 if bad_reply == 'unknown' and group.endswith('g0000') else 200,
+                )
+
+            def log_message(self, *_args):
+                pass
+
+        server = ThreadingHTTPServer(('127.0.0.1', 0), Ingress)
+        thread = threading.Thread(target=server.serve_forever)
+        thread.start()
+        try:
+            with tempfile.TemporaryDirectory() as folder:
+                root = Path(folder)
+                cases = root / 'cases.json'
+                cases.write_text(
+                    json.dumps(
+                        [
+                            {
+                                'name': 'group',
+                                'expect': 'TEST_OK',
+                                'http_group': {
+                                    'url': f'http://127.0.0.1:{server.server_port}/bots/test/sync',
+                                    'secret_env': 'LANGBOT_QA_GROUP_SECRET',
+                                    'message': 'ai Test group history',
+                                },
+                            }
+                        ]
+                    )
+                )
+                state = root / 'group-state.json'
+                command = [
+                    sys.executable,
+                    str(Path(__file__).with_name('regression_traffic.py')),
+                    '--lbctl',
+                    'unused',
+                    '--config',
+                    'unused',
+                    '--cases',
+                    str(cases),
+                    '--duration',
+                    '0.8',
+                    '--rps',
+                    '40',
+                    '--workers',
+                    '3',
+                    '--cooldown',
+                    '0',
+                    '--members-per-group',
+                    '2',
+                    '--health',
+                    f'core=http://127.0.0.1:{server.server_port}/healthz',
+                ]
+                env = {**os.environ, 'LANGBOT_QA_GROUP_SECRET': secret}
+                summaries = []
+                for groups in (4, 8):
+                    output = root / f'phase-{groups}'
+                    result = subprocess.run(
+                        [*command, '--groups', str(groups), '--group-state', str(state), '--output', str(output)],
+                        env=env,
+                        capture_output=True,
+                        text=True,
+                        timeout=15,
+                    )
+                    self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                    summaries.append(json.loads((output / 'summary.json').read_text()))
+                first = summaries[0]['groups']['sessions']
+                second = summaries[1]['groups']['sessions']
+                self.assertEqual(len(first), 4)
+                self.assertEqual(len(second), 8)
+                for group in first:
+                    self.assertGreater(second[group]['successful_turns_total'], first[group]['successful_turns_total'])
+                    self.assertEqual(len(second[group]['members_this_phase']), 2)
+                self.assertFalse(failures)
+                self.assertNotIn(secret, state.read_text())
+                bad_reply = True
+                output = root / 'failed'
+                result = subprocess.run(
+                    [*command, '--groups', '4', '--output', str(output)],
+                    env=env,
+                    capture_output=True,
+                    text=True,
+                    timeout=15,
+                )
+                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
+                summary = json.loads((output / 'summary.json').read_text())
+                failed = next(item for group, item in summary['groups']['sessions'].items() if group.endswith('g0000'))
+                self.assertTrue(failed['failed'])
+                self.assertEqual(failed['requests_this_phase'], 1)
+                bad_reply = 'unknown'
+                result = subprocess.run(
+                    [*command, '--groups', '4', '--output', str(root / 'unknown')],
+                    env=env,
+                    capture_output=True,
+                    text=True,
+                    timeout=15,
+                )
+                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
+                unknown = json.loads((root / 'unknown/summary.json').read_text())
+                self.assertEqual(unknown['status'].get('unknown'), 1)
+                interrupted = json.loads(state.read_text())
+                interrupted['in_progress'] = True
+                state.write_text(json.dumps(interrupted))
+                result = subprocess.run(
+                    [*command, '--groups', '4', '--group-state', str(state), '--output', str(root / 'interrupted')],
+                    env=env,
+                    capture_output=True,
+                    text=True,
+                    timeout=15,
+                )
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn('interrupted', result.stderr)
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join()
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/test_regression_job.py

@@ -164,15 +164,31 @@ class JobTests(unittest.TestCase):
             root = Path(folder)
             job = Job({}, profile(), root / 'profile.json', root)
             job.state['record'] = {'data_dir': str(job.deploy / 'data')}
+            containers = ['ours', 'other']
 
             def checked(args, timeout=900):
                 if args[1] == 'ps':
-                    return 'ours\nother\n'
-                source = str(job.deploy / 'data/box/tenant') if args[-1] == 'ours' else '/unrelated/box/tenant'
-                return json.dumps([{'Source': source}])
+                    return '\n'.join(containers)
+                return '\n'.join(
+                    json.dumps(
+                        {
+                            'Id': container,
+                            'Mounts': [
+                                {
+                                    'Source': str(job.deploy / 'data/box/tenant')
+                                    if container != 'other'
+                                    else '/unrelated/box/tenant'
+                                }
+                            ],
+                        }
+                    )
+                    for container in args[4:]
+                )
 
             job.checked = checked
             self.assertEqual(job.owned_sandboxes(), ['ours'])
+            containers.append('later')
+            self.assertEqual(job.owned_sandboxes(), ['ours', 'later'])
 
 
 if __name__ == '__main__':

tools/test_regression_suite.py

@@ -60,6 +60,87 @@ def profile():
 
 
 class ProfileTests(unittest.TestCase):
+    def test_created_bot_verification_failure_can_only_defer_bot_readback(self):
+        with tempfile.TemporaryDirectory() as folder:
+            root = Path(folder)
+            suite = Suite(profile(), root / 'profile.json', root)
+            response = {
+                'data': {'operation': 'bot.create', 'uuid': 'owned-bot', 'bot': {'uuid': 'owned-bot'}},
+                'error': {'type': 'verification_failed'},
+            }
+            result = subprocess.CompletedProcess([], 1, json.dumps(response), '')
+            with patch('regression_suite.subprocess.run', return_value=result):
+                with self.assertRaises(RuntimeError):
+                    suite.call(['bot', 'create'], {'name': 'QA'})
+                self.assertEqual(
+                    suite.call(['bot', 'create'], {'name': 'QA'}, defer_bot_readback=True)['uuid'], 'owned-bot'
+                )
+                with self.assertRaises(RuntimeError):
+                    suite.call(['pipeline', 'apply'], {}, defer_bot_readback=True)
+
+    def test_group_bot_is_owned_before_raw_readback_and_warmed_before_load(self):
+        with tempfile.TemporaryDirectory() as folder:
+            root = Path(folder)
+            current = profile()
+            current['resources'] = [
+                {
+                    'id': 'chat_bot',
+                    'kind': 'bot',
+                    'body': {
+                        'name': 'QA',
+                        'adapter': 'http_bot',
+                        'enable': True,
+                        'adapter_config': {'inbound_secret': '${group_secret}'},
+                        'event_bindings': [{'target_type': 'pipeline', 'target_uuid': 'pipeline-a'}],
+                    },
+                }
+            ]
+            current['cases'] = [
+                {
+                    'name': 'chat',
+                    'expect': 'OK',
+                    'http_group': {
+                        'url': 'http://localhost/bots/${chat_bot}/sync',
+                        'message': 'ai Test',
+                        'secret_env': 'LANGBOT_QA_GROUP_SECRET',
+                    },
+                }
+            ]
+            current['phases'][0]['groups'] = 1
+            validate(current)
+            for correct_route in (False, True):
+                suite = Suite(current, root / 'profile.json', root)
+
+                def call(args, body=None, **kwargs):
+                    if args == ['whoami']:
+                        return {'workspace_uuid': 'workspace-a'}
+                    if args == ['bot', 'create']:
+                        self.assertTrue(kwargs['defer_bot_readback'])
+                        return {'uuid': 'owned-bot'}
+                    raise AssertionError(args)
+
+                suite.call = call
+                actual = copy.deepcopy(current['resources'][0]['body'])
+                actual['adapter_config']['inbound_secret'] = '***'
+                if not correct_route:
+                    actual['event_bindings'][0]['target_uuid'] = 'wrong-pipeline'
+                with (
+                    patch.object(suite, 'read_bot', return_value=actual),
+                    patch('regression_suite.group_request', return_value={'code': 0}) as warmup,
+                ):
+                    if correct_route:
+                        suite.prepare()
+                        warmup.assert_called_once()
+                        self.assertNotIn(suite.variables['group_secret'], (root / 'resources.json').read_text())
+                    else:
+                        with self.assertRaisesRegex(RuntimeError, 'Bot readback differs'):
+                            suite.prepare()
+                        warmup.assert_not_called()
+                    self.assertEqual(
+                        json.loads((root / 'owned.json').read_text())['resources'],
+                        [{'kind': 'bot', 'uuid': 'owned-bot'}],
+                    )
+
     def test_normal_profile_and_overrides_contract(self):
         validate(json.loads(Path(__file__).with_name('regression_profiles').joinpath('normal.json').read_text()))
         bad = profile()