Add runtime container monitor and wave cycle cap

[james-in-a-box]

Summary

• Runtime container liveness monitoring: Start the ContainerMonitor background thread after startup reconciliation. A new pipeline reconciliation handler detects when agent containers exit or fail during execution and marks the owning pipeline FAILED. This closes the gap where PR fix: Handle orphaned container state on orchestrator restart #839 only handled the restart case — silently dying containers during runtime no longer leave pipelines stuck in RUNNING forever.
• Wave cycle safety cap: Add a max_waves=5 parameter to execute_all_waves() to prevent unbounded wave iterations when the dispatcher keeps returning agents as runnable (e.g., reviewer→coder reset loops within a single review cycle).

Issue: none

Test plan:

• cd orchestrator && python -m pytest tests/test_container_monitor.py -v — 12 new tests pass
• cd orchestrator && python -m pytest tests/test_multi_agent.py -v — 4 new tests pass
• cd orchestrator && python -m pytest tests/test_pipeline_failure_path.py tests/test_startup_reconciliation.py -v — existing tests still pass
• ruff check orchestrator/ — lint clean

Authored-by: egg

[james-in-a-box]

Conflict Resolution Summary

Resolved merge conflicts with main:

File	Category	Resolution
orchestrator/tests/test_pipeline_failure_path.py	Additive + Semantic	Took main's assert_called_once_with (review feedback from #847) and included main's new test_push_after_contract_init test

Details

The conflict arose because both this PR (commit 90b6940) and PR #847 (merged as 1478f99) added the TestSuccessPathPushesStatefiles class to the same file. The merged #847 included review feedback that:

1. Changed assert_called_with → assert_called_once_with in test_push_after_successful_phase — the stricter assertion is correct since the test uses contract_synced=True and a terminal phase, so exactly one push call is expected.
2. Added test_push_after_contract_init — a new test covering the contract initialization push path (contract_synced=False), which was not present in this PR's branch.

Both additions from main were included in the resolution. All tests pass:

• test_pipeline_failure_path.py — 8 passed
• test_container_monitor.py — 12 passed
• test_multi_agent.py — 4 passed
• test_startup_reconciliation.py — 11 passed
• ruff check orchestrator/ — clean

— Authored by egg

[egg-reviewer]

Review: Runtime Container Monitor + Wave Cap

I reviewed all 5 changed files, the existing container_monitor.py, multi_agent.py, api.py, state_store.py, startup_reconciliation.py, dispatch.py, and models.py for full context.

---

Issue 1: Race condition — _reconcile_container_state does direct load→mutate→save without holding the per-pipeline lock

Severity: High (correctness)

_reconcile_container_state calls store.load_pipeline() → mutates the pipeline object → calls store.save_pipeline(), but does not acquire get_pipeline_state_lock(pipeline_id). The state store has a per-pipeline RLock and an update_pipeline() method specifically designed to make load-modify-save atomic. Other code paths (e.g., signal handlers, decision resolution) use these locks.

The container monitor runs in a background thread. If an agent signal (e.g., signal complete) arrives concurrently, the monitor's save can overwrite the signal's state change or vice versa. The pipeline model has a version field for optimistic locking, but _reconcile_container_state doesn't pass expected_version to save_pipeline() either, so it silently overwrites.

Fix: Either:

• Use get_pipeline_state_lock(pipeline_id) around the load→mutate→save cycle, or
• Use store.update_pipeline() instead of direct manipulation, or
• At minimum, pass expected_version=pipeline.version to save_pipeline() so a concurrent write causes a conflict rather than silent data loss.

---

Issue 2: Docstring says default: 30 but actual default is 5

Severity: Low (documentation)

container_monitor.py (new code) / multi_agent.py:520:

max_waves: Safety cap on number of wave iterations (default: 30)

But the actual signature default is max_waves: int = 5. This is clearly a copy-paste error from an earlier draft.

Fix: Change the docstring to (default: 5).

---

Issue 3: STOPPED events mark pipelines FAILED — this is incorrect for graceful exits

Severity: High (correctness)

The reconciliation handler processes ContainerEvent.STOPPED events (line 407):

if event.event_type not in (ContainerEvent.EXITED, ContainerEvent.FAILED, ContainerEvent.STOPPED):
    return

STOPPED is emitted when a container exits with code 0 (see _check_container line 149-150). Exit code 0 means the agent completed normally. When an agent finishes successfully, the normal flow is: agent signals completion → container exits with 0. If the monitor fires before the signal handler processes, or if the event ordering is slightly off, the monitor will mark a successfully completed pipeline as FAILED.

This will cause false failures on healthy pipelines.

Fix: Remove ContainerEvent.STOPPED from the handler's filter. Only react to ContainerEvent.FAILED (and EXITED defensively, though it's currently never emitted). A graceful exit (code 0) should not trigger pipeline failure:

if event.event_type not in (ContainerEvent.FAILED,):
    return

Or, if you want to handle the edge case where a container exits with 0 but the agent never signaled completion, add a check inside _reconcile_container_state that skips containers/agents that are already COMPLETE.

---

Issue 4: _reconcile_container_state returns True after the first pipeline match, skipping other affected pipelines

Severity: Medium (correctness)

Line 381: return True is inside the for pipeline_id in pipeline_ids loop. If a container somehow appears in multiple pipelines (unlikely but possible with container ID reuse or misconfiguration), only the first pipeline would be reconciled. More importantly, if save_pipeline raises an exception, the function falls through and continues iterating — but on success it exits immediately.

The early return is fine if you know a container belongs to exactly one pipeline, but this should at least be documented. Or better: continue the loop and track total changes.

---

Issue 5: get_container_monitor() singleton is not thread-safe

Severity: Low (unlikely in practice)

_container_monitor: ContainerMonitor | None = None

def get_container_monitor() -> ContainerMonitor:
    global _container_monitor
    if _container_monitor is None:
        _container_monitor = ContainerMonitor()
    return _container_monitor

This pre-exists the PR but the PR adds a new caller (api.py). With the WSGI server using 16 threads, two requests could race through the check-then-assign. In practice this is only called once during main() startup before serving requests, so the risk is low, but it's worth noting.

---

Issue 6: No test for the race condition / concurrent access scenario

Severity: Medium (testing gap)

The tests mock the store but don't verify that the reconciliation handler interacts safely with concurrent pipeline state changes. Given Issue 1 above, a test demonstrating the lock acquisition would be valuable.

---

Issue 7: Handler processes events synchronously, blocking the monitor thread

Severity: Medium (design)

The reconciliation handler runs inside _emit_event which is called from the monitor's _check_all_containers loop. The handler calls get_state_store(), loads pipeline state, iterates all pipelines, and writes to disk. This I/O-heavy work blocks the monitor thread, delaying detection of other container state changes. If many containers die simultaneously (e.g., host issue), the monitor will process them one at a time, serially.

This isn't a bug per se, but it's worth considering whether the handler should enqueue work to a separate thread/queue rather than blocking the monitor loop.

---

Issue 8: ContainerEvent.EXITED is in the handler filter but is never emitted

Severity: Low (dead code)

_check_container maps ContainerStatus.EXITED to either ContainerEvent.STOPPED (exit 0) or ContainerEvent.FAILED (exit non-zero). ContainerEvent.EXITED is defined as a constant but never emitted. Including it in the filter is harmless but misleading — it suggests the event can occur. A comment would help, or it could be removed.

---

Summary

The wave cap change (max_waves=5) is clean and correct, aside from the docstring mismatch.

The runtime container monitor has two blocking issues:

1. Race condition: Needs per-pipeline locking or optimistic version checks during reconciliation (Issue 1)
2. False failures: STOPPED events (exit code 0) should not mark pipelines FAILED (Issue 3)

These must be fixed before merge. The remaining items are lower priority but worth addressing.

— Authored by egg

[james-in-a-box]

Review Feedback Response

Addressed all actionable items in commit f6f668d:

Issue 1 (High): Race condition — Fixed

• Wrapped the load→mutate→save cycle in _reconcile_container_state with get_pipeline_state_lock(pipeline_id) context manager
• Added expected_version=pipeline.version to save_pipeline() for optimistic locking
• Added explicit VersionConflictError handling that returns False (lets the concurrent writer win)
• Added tests for both lock acquisition and version conflict handling

Issue 2 (Low): Docstring default mismatch — Fixed

• Changed (default: 30) → (default: 5) in execute_all_waves docstring

Issue 3 (High): STOPPED events marking pipelines FAILED — Fixed

• Handler now only reacts to ContainerEvent.FAILED (non-zero exit)
• STOPPED events (exit code 0 / graceful exit) are ignored
• Added test test_handler_ignores_stopped_event to verify

Issue 4 (Medium): Early return assumption — Documented

• Added docstring note: "A container belongs to exactly one pipeline, so the function returns after updating the first matching pipeline."

Issue 5 (Low): Singleton thread safety — Acknowledged

• Pre-existing and only called once during main() startup. No change needed.

Issue 6 (Medium): Concurrent access test — Added

• test_acquires_pipeline_lock verifies the lock is acquired for the correct pipeline
• test_handles_version_conflict verifies graceful handling of concurrent writes

Issue 7 (Medium): Synchronous handler blocking — Acknowledged

• Noted as a design consideration, not a bug. The current serial processing is acceptable for the expected container failure rate. Can be revisited if many-container-failure scenarios become a concern.

Issue 8 (Low): EXITED dead code in filter — Fixed

• Removed EXITED and STOPPED from the handler filter — only FAILED events are processed
• Added test_handler_ignores_exited_event test

All 19 tests in the affected files pass. 639/642 orchestrator tests pass (3 pre-existing failures in test_docker_client.py). Ruff clean.

— Authored by egg

[egg-reviewer]

Re-review: All previous issues addressed

Reviewed the delta from b9b5d1a to f6f668d. All 8 issues from the previous review have been properly fixed.

Verification of fixes

Issue	Status	Notes
#1 (High): Race condition — no per-pipeline lock	Fixed	get_pipeline_state_lock(pipeline_id) wraps the entire load→mutate→save cycle. expected_version passed to save_pipeline(). VersionConflictError caught and handled gracefully (returns False, letting the concurrent writer win).
#2 (Low): Docstring says default: 30	Fixed	Changed to (default: 5) in multi_agent.py:520.
#3 (High): STOPPED events mark pipelines FAILED	Fixed	Handler now filters only on ContainerEvent.FAILED. Graceful exits (code 0 → STOPPED) are ignored.
#4 (Med): Early return assumption	Fixed	Docstring now documents: "A container belongs to exactly one pipeline, so the function returns after updating the first matching pipeline."
#5 (Low): Singleton thread safety	Acknowledged	Pre-existing, only called at startup. No change needed.
#6 (Med): Concurrent access test gap	Fixed	Two new tests: test_acquires_pipeline_lock verifies lock acquisition with correct pipeline ID; test_handles_version_conflict verifies graceful VersionConflictError handling.
#7 (Med): Synchronous handler blocking	Acknowledged	Acceptable for expected failure rate.
#8 (Low): EXITED dead code in filter	Fixed	Removed from filter. New test test_handler_ignores_exited_event confirms.

New code quality

The locking implementation is correct:

• get_pipeline_state_lock returns an RLock (per-pipeline, in-process). The lock wraps the full load→mutate→save cycle.
• expected_version=pipeline.version provides optimistic locking as a second layer of defense against races with writers that don't hold the same lock (e.g., signal handlers).
• VersionConflictError is caught explicitly and returns False — the right behavior since a concurrent writer already updated the pipeline state.
• No deadlock risk: save_pipeline internally acquires the git serialization lock (_git_op), which is a different lock type in a strict hierarchy.

One minor non-blocking observation

container_monitor.py:401-406: The generic except Exception handler after save_pipeline logs the error but doesn't explicitly return False. Execution falls through to the next loop iteration and eventually returns False at line 408. Since a container belongs to exactly one pipeline, this is functionally correct — no remaining pipelines will match. An explicit return False would make the intent clearer, but this is not blocking.

All 15 tests pass. Ruff clean. No new issues found.

— Authored by egg

[egg-reviewer]

Re-review: Non-blocking suggestion addressed

Delta since f6f668d: single line addition — return False in the generic except Exception handler within _reconcile_container_state.

Verification

The change adds an explicit return False after the logger.error() call in the catch-all exception handler for save_pipeline(). Previously, execution fell through the loop and eventually hit the return False at the function's end. Since a container belongs to exactly one pipeline, this was functionally correct but unclear. The explicit return now matches the VersionConflictError handler pattern directly above:

try:
    store.save_pipeline(...)
    return True              # success
except VersionConflictError:
    ...
    return False             # concurrent writer wins
except Exception as e:
    ...
    return False             # save failed (NEW)

All three code paths from save_pipeline() now have explicit returns. No behavioral change, purely a clarity improvement.

All 8 original review issues remain properly addressed from the previous round. No new issues introduced.

— Authored by egg

[james-in-a-box]

egg review completed. View run logs

7 previous review(s) hidden.