| phase | quick-260707-g84 | |||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| plan | 1 | |||||||||||||||||||||||||||||||||||||||||||
| type | execute | |||||||||||||||||||||||||||||||||||||||||||
| wave | 1 | |||||||||||||||||||||||||||||||||||||||||||
| depends_on | ||||||||||||||||||||||||||||||||||||||||||||
| files_modified |
|
|||||||||||||||||||||||||||||||||||||||||||
| autonomous | true | |||||||||||||||||||||||||||||||||||||||||||
| requirements |
|
|||||||||||||||||||||||||||||||||||||||||||
| must_haves |
|
Purpose: make an explicit WORKER_MAX_JOBS cap authoritative in lane mode (as a
ceiling) so a single-lane, memory-constrained host is protected, and belt-and-braces
pin the analyze lane to 1 in the cloud-agent compose.
Output: corrected resolution logic + startup log, compose knob, two tests, doc note.
<execution_context> @$HOME/.claude/get-shit-done/workflows/execute-plan.md @$HOME/.claude/get-shit-done/templates/summary.md </execution_context>
@.planning/STATE.md @./CLAUDE.mdsrc/phaze/config.py (BaseSettings, env_file=".env", NO env_prefix — so field FOO binds to env FOO): worker_max_jobs: int = 8 # env WORKER_MAX_JOBS lane_analyze_concurrency: int = Field(default=4, validation_alias=AliasChoices("PHAZE_LANE_ANALYZE_CONCURRENCY", "lane_analyze_concurrency")) lane_fingerprint_concurrency: int = Field(default=2, ...) # PHAZE_LANE_FINGERPRINT_CONCURRENCY lane_meta_concurrency: int = Field(default=2, ...) # PHAZE_LANE_META_CONCURRENCY lane_io_concurrency: int = Field(default=4, ...) # PHAZE_LANE_IO_CONCURRENCY
src/phaze/tasks/agent_worker.py (module-level resolution, ~lines 300-351): _LANE_CONCURRENCY_ATTR: dict[str, str] = {"analyze": "lane_analyze_concurrency", "fingerprint": "lane_fingerprint_concurrency", "meta": "lane_meta_concurrency", "io": "lane_io_concurrency"} _lane = os.environ.get("PHAZE_AGENT_LANE") or None _settings_obj = get_settings()
logger = structlog.get_logger(name) # existing module logger
startup() already emits logger.info("phaze.tasks.agent_worker startup complete ...") after configure_logging()
tests/agents/tasks/test_agent_worker_lanes.py (reload harness — reads mod.settings["concurrency"], NO worker boot): _reload_worker(monkeypatch, *, lane: str | None) -> ModuleType # sets base agent env + PHAZE_AGENT_LANE, evicts+reimports _set_base_env(monkeypatch) # does NOT set WORKER_MAX_JOBS (so default 8 applies) autouse fixture _evict_agent_worker evicts phaze.tasks.agent_worker after each test
tests/agents/deployment/test_cloud_agent_compose.py: _load_cloud_agent_compose() -> dict # yaml.safe_load, no interpolation _env_to_strs(env) -> list[str] # normalizes environment to ["KEY=VALUE", ...]
Task 1: Make worker_max_jobs a ceiling in lane-mode concurrency resolution + startup log src/phaze/tasks/agent_worker.py, tests/agents/tasks/test_agent_worker_lanes.py - lane=analyze, WORKER_MAX_JOBS unset (default 8): effective concurrency == 4 (lane default, unclamped) — file-server case unchanged. - lane=analyze, WORKER_MAX_JOBS=1: effective concurrency == 1 (lane 4 clamped by ceiling) — the OCI A1 compute-agent fix. - all-mode (PHAZE_AGENT_LANE unset): effective concurrency == worker_max_jobs (existing test_all_mode_preserves_todays_behavior still passes). - Existing test_lane_selects_queue_functions_and_concurrency still passes (default worker_max_jobs=8 >= every lane default, so min() == lane value). In the module-level resolution block (the `if _lane is not None:` branch, ~line 320), change the lane-mode concurrency to clamp by the worker_max_jobs ceiling: resolve `_concurrency` as `min(getattr(_settings_obj, _LANE_CONCURRENCY_ATTR[_lane]), _settings_obj.worker_max_jobs)`. Preserve the all-mode branch (`_concurrency = _settings_obj.worker_max_jobs`) unchanged. Capture the raw lane knob value and a clamped boolean at module scope (e.g. `_lane_raw_concurrency` and `_concurrency_clamped = _lane is not None and _concurrency < _lane_raw_concurrency`) so the startup log can report whether the ceiling bit. Add a one-line effective-concurrency log inside the existing `startup()` hook, AFTER `configure_logging(...)` (so it renders through the configured pipeline) and matching the module's existing `logger.info("phaze.tasks.agent_worker ...", ...)` structlog style: report the effective concurrency, the lane (or ""), and whether the worker_max_jobs ceiling clamped it. Do NOT log at module-import time (logging is not yet configured there). Add a short comment tagging this as the quick-260707-g84 memory-safety ceiling and noting WORKER_MAX_JOBS is a ceiling in lane mode. Do NOT use fenced code or add new imports (structlog logger already exists). Then add unit tests to tests/agents/tasks/test_agent_worker_lanes.py using the existing `_reload_worker` harness: one test setting `monkeypatch.setenv("WORKER_MAX_JOBS", "1")` before reload with lane="analyze" asserting `mod.settings["concurrency"] == 1`, and one asserting the default (no WORKER_MAX_JOBS override) lane="analyze" resolves to 4. Follow the file's existing style (from __future__, importlib reload, get_settings import inside the test body). uv run pytest tests/agents/tasks/test_agent_worker_lanes.py -x -q && uv run ruff check src/phaze/tasks/agent_worker.py tests/agents/tasks/test_agent_worker_lanes.py && uv run mypy src/phaze/tasks/agent_worker.py Lane-mode concurrency = min(lane knob, worker_max_jobs); WORKER_MAX_JOBS=1 + lane=analyze yields 1; default yields 4; startup log reports effective concurrency + clamp; all existing lane tests still pass; ruff + mypy clean. Task 2: Pin analyze lane to 1 in cloud-agent compose + guard test docker-compose.cloud-agent.yml, tests/agents/deployment/test_cloud_agent_compose.py In docker-compose.cloud-agent.yml under `services.worker.environment`, add `PHAZE_LANE_ANALYZE_CONCURRENCY=1` immediately alongside the existing lane env, with a comment noting that in lane mode the lane knob is what actually governs a lane worker's concurrency (WORKER_MAX_JOBS is only a ceiling — quick-260707-g84). Keep the existing `PHAZE_AGENT_LANE=analyze` line. Do not touch other services/volumes. Then in tests/agents/deployment/test_cloud_agent_compose.py add a test asserting the compute worker caps analyze-lane concurrency to 1: parse the worker environment via the existing `_env_to_strs(...)` helper and assert `PHAZE_LANE_ANALYZE_CONCURRENCY=1` is present (accept the `<= WORKER_MAX_JOBS` framing in the docstring). Follow the file's existing test/docstring conventions and reference quick-260707-g84. uv run pytest tests/agents/deployment/test_cloud_agent_compose.py -x -q && uv run ruff check tests/agents/deployment/test_cloud_agent_compose.py docker-compose.cloud-agent.yml sets PHAZE_LANE_ANALYZE_CONCURRENCY=1 with an explanatory comment; new compose-guard test asserts it and passes; existing compose tests still pass. Task 3: Document lane-knob-governs / WORKER_MAX_JOBS-is-a-ceiling semantics docs/agent-queue-lanes.md Add a short note (in the "Compute (cloud/x86) agent — single lane" section and/or the core-budget rationale) stating that in lane mode the per-lane concurrency knob (`PHAZE_LANE__CONCURRENCY`) governs the worker's concurrency, and `WORKER_MAX_JOBS` now acts only as a CEILING (`concurrency = min(lane knob, worker_max_jobs)`). Call out the memory-safety motivation: on the OCI Ampere A1 (12 GB) compute agent, `process_file` peaks ~8 GB, so the analyze lane is pinned to 1 (`PHAZE_LANE_ANALYZE_CONCURRENCY=1`) — setting only `WORKER_MAX_JOBS=1` is inert in lane mode. Tag the note quick-260707-g84. Keep the existing gsd:doc marker on line 1 intact. grep -q 'PHAZE_LANE_ANALYZE_CONCURRENCY' docs/agent-queue-lanes.md && grep -qi 'ceiling' docs/agent-queue-lanes.md docs/agent-queue-lanes.md explains lane knob governs + WORKER_MAX_JOBS is a ceiling in lane mode, with the A1 memory-safety rationale. Full gate for the quick task: - `uv run pytest tests/agents/tasks/test_agent_worker_lanes.py tests/agents/deployment/test_cloud_agent_compose.py -q` - `uv run ruff check .` - `uv run mypy .` - Spot: reload agent_worker with PHAZE_AGENT_LANE=analyze + WORKER_MAX_JOBS=1 resolves concurrency 1; default resolves 4; all-mode resolves worker_max_jobs.<success_criteria>
- Lane-mode concurrency =
min(lane knob, worker_max_jobs); explicit WORKER_MAX_JOBS cap is authoritative. - Cloud-agent compose pins analyze lane to 1 via PHAZE_LANE_ANALYZE_CONCURRENCY=1.
- Effective concurrency (+ lane + clamp) is logged at startup.
- New unit + compose-guard tests pass; existing lane/compose tests unaffected.
- Docs updated. ruff + mypy + pytest green (90% coverage maintained). </success_criteria>