Document batch worker flows and harden lock-contention coverage

Author: ddy6

README.md

@@ -14,7 +14,8 @@ For expensive black-box objectives, Looptimum starts with bounded exploration
 and then shifts to surrogate-guided suggestion ranking to reduce wasted trials.
 Its key differentiator is operational: a file-backed, resumable workflow that
 keeps state and decision trace local, which fits restricted and client-controlled
-environments. The usage model stays simple (`suggest -> evaluate -> ingest`);
+environments. The usage model stays simple (`suggest -> evaluate -> ingest`,
+with optional locked batches);
 see [`docs/how-it-works.md`](docs/how-it-works.md) for algorithm behavior and
 tuning consequences.
 For a spec-style contract summary, use
@@ -43,7 +44,7 @@ For a spec-style contract summary, use
 Looptimum replaces ad hoc sweep loops with a small, explicit workflow:
 
 1. Define parameter bounds, objective schema, and optional constraints.
-2. `suggest` one trial.
+2. `suggest` one trial by default, or allocate a locked batch with `--count N`.
 3. Run that trial in your environment.
 4. `ingest` the result and repeat.
 
@@ -228,12 +229,23 @@ expanded stub in
 
 ### `suggest` Output
 
+Count `1` keeps the historical single-suggestion payload. Count `> 1` emits a
+bundle JSON object by default:
+
+- `schema_version`
+- `count`
+- `suggestions` (array of canonical suggestion payloads)
+
+Use `--jsonl` to emit one canonical suggestion JSON object per line for worker
+handoff.
+
 Each suggestion includes:
 
 - `schema_version` (semver string, emitted by runtime)
 - `trial_id`
 - `params`
 - `suggested_at`
+- `lease_token` (only when `worker_leases.enabled` is true)
 
 ### `ingest` Required Fields
 
@@ -284,9 +296,7 @@ Best ranking rule:
 
 ### Compatibility Notes
 
-- `success` is accepted as a deprecated alias and normalized to `ok`.
-- Legacy `failure_reason` is accepted as a deprecated alias and normalized to
-  `terminal_reason`.
+- Canonical statuses are `ok`, `failed`, `killed`, and `timeout`.
 - For non-`ok` outcomes with no reason provided, ingest synthesizes
   `terminal_reason` as `status=<status>`.
 - `v0.2.x` state without `schema_version` (or with `0.2.x`) upgrades in-memory
@@ -319,6 +329,13 @@ Best ranking rule:
 - `validate [--strict]`: sanity-check config/state; warnings are non-fatal unless `--strict`.
 - `doctor [--json]`: print environment/backend/state diagnostics.
 
+Lease note:
+
+- when `worker_leases.enabled` is true, `suggest` emits `lease_token` and
+  workers must echo it on `heartbeat` and `ingest`
+- `max_pending_trials`, when configured, rejects the whole requested batch
+  before any pending state is created
+
 ## Templates (Choose Your Starting Level)
 
 ### Template Matrix (Feature Parity + Intended Use)
@@ -342,6 +359,8 @@ The `examples/` folder shows integration patterns, not benchmark leaderboards.
   objective (`suggest -> evaluate -> ingest -> status`, typically under one minute)
 - `docs/examples/multi_objective/`: generated multi-objective report/state pack
   with weighted-sum and lexicographic objective-schema examples
+- `docs/examples/batch_async/`: batch bundle, JSONL handoff, lease-token, and
+  pending-state example pack
 
 Run the tiny end-to-end objective from repo root:
 
@@ -447,3 +466,15 @@ python3 templates/bo_client_demo/run_bo.py suggest \
   --project-root templates/bo_client_demo \
   --json-only
 ```
+
+For worker fan-out, use line-delimited output:
+
+```bash
+python3 templates/bo_client_demo/run_bo.py suggest \
+  --project-root templates/bo_client_demo \
+  --count 3 \
+  --jsonl
+```
+
+Bundle JSON, JSONL handoff, `max_pending_trials`, and lease-token examples are
+captured in `docs/examples/batch_async/README.md`.

client_harness_template/tests/test_phase6_assets.py

@@ -11,6 +11,7 @@
 PYPROJECT = REPO_ROOT / "pyproject.toml"
 TYPE_SAFETY_DOC = REPO_ROOT / "docs" / "type-safety.md"
 MULTI_OBJECTIVE_EXAMPLE = REPO_ROOT / "docs" / "examples" / "multi_objective"
+BATCH_ASYNC_EXAMPLE = REPO_ROOT / "docs" / "examples" / "batch_async"
 
 
 def test_golden_acquisition_log_has_expected_shape_and_timestamps() -> None:
@@ -109,3 +110,67 @@ def test_multi_objective_example_pack_has_expected_artifacts() -> None:
     manifest_payload = json.loads(manifest_path.read_text(encoding="utf-8"))
     assert manifest_payload["scalarization_policy"] == "weighted_sum"
     assert manifest_payload["objective_vector"] == {"loss": 0.3, "throughput": 2.0}
+
+
+def test_batch_async_example_pack_has_expected_artifacts() -> None:
+    assert BATCH_ASYNC_EXAMPLE.exists(), f"missing batch/async example pack: {BATCH_ASYNC_EXAMPLE}"
+
+    readme_path = BATCH_ASYNC_EXAMPLE / "README.md"
+    config_path = BATCH_ASYNC_EXAMPLE / "bo_config.json"
+    bundle_path = BATCH_ASYNC_EXAMPLE / "suggestion_bundle.json"
+    jsonl_path = BATCH_ASYNC_EXAMPLE / "suggestions.jsonl"
+    status_suggest_path = BATCH_ASYNC_EXAMPLE / "status_after_batch_suggest.json"
+    status_ingest_path = BATCH_ASYNC_EXAMPLE / "status_after_ingest.json"
+    report_path = BATCH_ASYNC_EXAMPLE / "state" / "report.json"
+    manifest_path = BATCH_ASYNC_EXAMPLE / "state" / "trials" / "trial_1" / "manifest.json"
+
+    for path in (
+        readme_path,
+        config_path,
+        bundle_path,
+        jsonl_path,
+        status_suggest_path,
+        status_ingest_path,
+        report_path,
+        manifest_path,
+    ):
+        assert path.exists(), f"missing batch/async example artifact: {path}"
+
+    config_payload = json.loads(config_path.read_text(encoding="utf-8"))
+    assert config_payload["batch_size"] == 2
+    assert config_payload["max_pending_trials"] == 3
+    assert config_payload["worker_leases"] == {"enabled": True}
+
+    bundle_payload = json.loads(bundle_path.read_text(encoding="utf-8"))
+    assert bundle_payload["count"] == 2
+    assert [item["trial_id"] for item in bundle_payload["suggestions"]] == [1, 2]
+    assert all(item["lease_token"] for item in bundle_payload["suggestions"])
+
+    jsonl_payloads = [
+        json.loads(line)
+        for line in jsonl_path.read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+    assert [item["trial_id"] for item in jsonl_payloads] == [1, 2]
+    assert [item["lease_token"] for item in jsonl_payloads] == [
+        item["lease_token"] for item in bundle_payload["suggestions"]
+    ]
+
+    suggest_status = json.loads(status_suggest_path.read_text(encoding="utf-8"))
+    assert suggest_status["pending"] == 2
+    assert suggest_status["leased_pending"] == 2
+    assert suggest_status["worker_leases_enabled"] is True
+
+    ingest_status = json.loads(status_ingest_path.read_text(encoding="utf-8"))
+    assert ingest_status["pending"] == 0
+    assert ingest_status["leased_pending"] == 0
+    assert ingest_status["best"]["trial_id"] == 2
+
+    report_payload = json.loads(report_path.read_text(encoding="utf-8"))
+    assert report_payload["counts"]["observations"] == 2
+    assert report_payload["top_trials"][0]["trial_id"] == 2
+
+    manifest_payload = json.loads(manifest_path.read_text(encoding="utf-8"))
+    assert manifest_payload["lease_token"] == bundle_payload["suggestions"][0]["lease_token"]
+    assert manifest_payload["heartbeat_count"] == 1
+    assert manifest_payload["heartbeat_meta"] == {"worker": "worker-1", "queue": "batch"}

docs/README.md

@@ -46,6 +46,7 @@ Reference artifacts:
 
 - `../PILOT.md`
 - `examples/README.md`
+- `examples/batch_async/README.md`
 - `examples/multi_objective/README.md`
 - `examples/state_snapshots/README.md`
 - `examples/decision_trace/README.md`

docs/ci-knob-tuning.md

@@ -81,8 +81,12 @@ Why unsupported:
 Queueing guidance:
 
 - bound in-flight evaluations using explicit concurrency limit
+- keep `max_pending_trials` aligned with the largest controller batch you plan
+  to request
 - ingest results in deterministic order (by completion time or trial id)
 - on repeated lock contention, fail fast and retry controller job
+- if `worker_leases.enabled` is on, preserve `lease_token` with each worker task
+  and echo it on `heartbeat` / `ingest`
 
 ## 4) Contamination Controls
 
@@ -144,7 +148,7 @@ jobs:
         with:
           python-version: "3.12"
       - run: python -m pip install -r requirements-dev.txt
-      - run: python templates/bo_client/run_bo.py suggest --project-root templates/bo_client --json-only > /tmp/suggestion.json
+      - run: python templates/bo_client/run_bo.py suggest --project-root templates/bo_client --count 3 --jsonl > /tmp/suggestions.jsonl
       - uses: actions/upload-artifact@v4
         with:
           name: looptimum-state
@@ -163,7 +167,7 @@ jobs:
         with:
           name: looptimum-state
           path: templates/bo_client/state
-      - run: echo "Run external evaluator here; write result payload artifact"
+      - run: echo "Run one worker per JSONL line; preserve lease_token if present and write one result payload artifact per trial"
 
   controller_ingest:
     needs: evaluator

docs/decision-trace.md

@@ -9,7 +9,8 @@ Decision traces are written as JSON Lines:
 
 - `state/acquisition_log.jsonl`
 
-Each line corresponds to one `suggest` decision attempt.
+Each line corresponds to one trial-level `suggest` decision attempt. Batch
+`suggest --count N` writes one line per allocated trial.
 
 Related runtime log:
 
@@ -58,6 +59,8 @@ Top-level fields:
 Important nuance:
 
 - successful `suggest` attempts create pending state for that `trial_id`
+- batched successful `suggest` commands create one such record per allocated
+  trial id
 - all-infeasible `suggest` attempts still log a decision, but they do not
   create a pending trial or increment authoritative state
 

docs/examples/README.md

@@ -10,6 +10,8 @@ Integration pattern note:
 
 Included:
 
+- `batch_async/`: generated bundle JSON, JSONL handoff, lease-token, and
+  state/report examples for batch suggest flows
 - `multi_objective/`: generated weighted-sum / lexicographic example pack with
   `status`, `report`, and trial-manifest outputs
 - `state_snapshots/`: sample state/log/CSV snapshots captured from a temp run

docs/examples/batch_async/README.md

@@ -0,0 +1,40 @@
+# Batch + Async Worker Example Pack
+
+Generated reference artifacts for a clean batch run captured from
+`templates/bo_client_demo`.
+
+This pack shows the current batch/async surface:
+
+- `bo_config.json`: example config with `batch_size = 2`,
+  `max_pending_trials = 3`, and `worker_leases.enabled = true`
+- `suggestion_bundle.json`: default count-2 `suggest` bundle output
+- `suggestions.jsonl`: equivalent worker-handoff JSONL form of the same
+  allocation
+- `status_after_batch_suggest.json`: pending-state headline with
+  `leased_pending = 2`
+- `result_1.json` / `result_2.json`: canonical ingest payloads for the leased
+  trials
+- `status_after_ingest.json`: final run headline after both leased ingests
+- `state/bo_state.json`: authoritative state after the batch completes
+- `state/report.json` / `state/report.md`: explicit report outputs for the run
+- `state/trials/trial_<id>/manifest.json`: per-trial manifests showing
+  `lease_token`, heartbeat metadata, and artifact pointers
+
+The captured flow is:
+
+1. controller runs `suggest --count 2 --json-only`
+2. workers receive either the bundle payload or the JSONL lines
+3. worker 1 claims its trial with `heartbeat --lease-token <token>`
+4. both results are ingested with matching `--lease-token`
+5. `report` is generated from the finished state
+
+Operational notes illustrated here:
+
+- count-1 `suggest` stays backward compatible; batch output starts at count `> 1`
+- `--jsonl` is the worker-oriented serialization for the same canonical
+  suggestion objects
+- `max_pending_trials` is a whole-batch guardrail, not a partial allocator
+- lease tokens are CLI-side claim checks; they are not embedded into ingest
+  payload JSON
+
+These files are documentation examples, not benchmark claims.

docs/examples/batch_async/bo_config.json

@@ -0,0 +1,41 @@
+{
+  "seed": 17,
+  "max_trials": 40,
+  "initial_random_trials": 6,
+  "candidate_pool_size": 600,
+  "surrogate": {
+    "type": "rbf_proxy",
+    "length_scale": 0.22
+  },
+  "acquisition": {
+    "type": "ucb",
+    "kappa": 1.8,
+    "xi": 0.01
+  },
+  "feature_flags": {
+    "enable_botorch_gp": false,
+    "fallback_to_proxy_if_unavailable": true,
+    "enable_service_api_preview": false,
+    "enable_dashboard_preview": false,
+    "enable_auth_preview": false
+  },
+  "batch_size": 2,
+  "max_pending_trials": 3,
+  "worker_leases": {
+    "enabled": true
+  },
+  "stopping": {
+    "min_improvement": 0.0001,
+    "patience_trials": 10
+  },
+  "paths": {
+    "state_file": "state/bo_state.json",
+    "observations_csv": "state/observations.csv",
+    "acquisition_log_file": "state/acquisition_log.jsonl",
+    "constraints_schema_file": "../_shared/schemas/constraints.schema.json",
+    "ingest_schema_file": "../_shared/schemas/ingest_payload.schema.json",
+    "objective_schema_schema_file": "../_shared/schemas/objective_schema.schema.json",
+    "search_space_schema_file": "../_shared/schemas/search_space.schema.json",
+    "suggestion_schema_file": "../_shared/schemas/suggestion_payload.schema.json"
+  }
+}

docs/examples/batch_async/objective_schema.json

@@ -0,0 +1,9 @@
+{
+  "primary_objective": {
+    "name": "loss",
+    "direction": "minimize",
+    "tolerance": 0.0,
+    "failure_handling": "record_and_continue"
+  },
+  "secondary_objectives": []
+}

docs/examples/batch_async/parameter_space.json

@@ -0,0 +1,16 @@
+{
+  "parameters": [
+    {
+      "name": "x1",
+      "type": "float",
+      "bounds": [0.0, 1.0],
+      "description": "First controllable factor"
+    },
+    {
+      "name": "x2",
+      "type": "float",
+      "bounds": [0.0, 1.0],
+      "description": "Second controllable factor"
+    }
+  ]
+}