refactor: isolate worker stdio and stream tool logs - #210
Open
Emin017 wants to merge 13 commits into
Open
Conversation
Worker response correlation now matches by request_id; notifications are queued separately. Process-group cleanup caches pgid at start and signals the group even after leader exits. Operation-scoped repair only touches the named step. Production stdio_server.main() installs StdioIsolation permanently; per-handler redirect_stdout_to_stderr removed from rpc_dispatch (now a no-op under permanent isolation). LogStreamReader archives unknown marker events as raw data instead of silently discarding them. Archive I/O errors are surfaced through state.error.
…repair, and marker matching Response correlation: separate id-keyed pending-response store from notification queue. read_response checks pending store before reading stdout. All decoded messages in every batch are preserved. Process-group escalation: check group liveness (killpg signal 0) after each signal+wait. Continue SIGTERM/SIGKILL even when proc.wait() has already returned but the group still has live members. Flow repair: require active_step (no unscoped fallback). Check json_write return value and raise OSError on write failure. Log stream: track active step/tool in state. Only a matching end marker closes the archive. Mismatched end markers are archived as raw data without changing state.
…machine markers, and emit from EngineFlow Response envelope validation: require jsonrpc=="2.0" and exactly one of result or error (with code+message) before storing or returning a response. Invalid envelopes raise WorkerProcessError. Process-group escalation: replace proc.wait()-based escalation with group-liveness polling via os.killpg(pgid, 0) with deadlines. After each signal, wait for the group to exit before escalating. Descendants that handle SIGTERM gracefully are not SIGKILL'd. Log stream state machine: only accept begin markers while inactive (no active step). A begin while active is archived as raw data without state change. Close archive handle properly on write failure. Step marker emission: EngineFlow.run_step() emits begin/end markers on stderr around tool execution. Begin after Ongoing persistence, end in finally path.
Worker: - Reap leader during group-liveness waits (zombie no longer keeps group visible; graceful shutdown completes before forceful deadline) - Validate response id as int|str|None; reject booleans and non-scalars Flow: - Move marker end after ALL step finalization (metrics, state persist, layout, DB cleanup, observer) via outer try/finally LogStream: - Archive close exception safety: close in finally regardless of flush New: - worker_operation.py: typed RunOperation orchestrator integrating WorkerClient + LogStreamReader + repair into OperationResult - Crash path: terminate group, drain reader, repair flow.json, return typed failure Tests: - Elapsed-time regression proving graceful terminate < 5s - Response id=true and id=[1] rejection - Operation orchestrator: success, RPC error, crash+repair, archive
Replace the fake one-RPC wrapper with the canonical session sequence: rpc.hello → workspace.open → flow.run → rpc.shutdown → EOF wait. Key changes: - Default argv now launches `ecc rpc serve --stdio --persistent-db` - Graceful shutdown via rpc.shutdown RPC, not signals - Archive completion is a required condition for success (archive error, reader timeout, or unmatched begin marker all force failure) - Protocol failures (dead worker, invalid envelope, EOF) route through crash recovery with flow.json repair - Non-object JSON marker payloads no longer crash parse_marker - LogStreamReader gains a `completed` property for checked drain Tests rewritten to exercise the full multi-request lifecycle with hello/open/run/shutdown, plus crash repair, protocol failure recovery, archive error detection, and non-object marker resilience.
…down - Send `directory` (not `path`) in workspace.open params to match WorkspaceOpenRequest schema - Extract workspaceId from open response and inject it into subsequent flow request params as workspace_id - Validate rpc.shutdown response (require result.ok is True) before waiting for process exit - Add real-server lifecycle tests proving hello/open/shutdown through the installed ecc rpc serve --stdio --persistent-db binary - Add contract tests asserting directory field and workspace_id injection
Wire the non-interactive `ecc run` path through the isolated worker process (RunOperation → flow.run RPC) instead of calling EngineFlow.run_steps() directly in-process. Falls back to direct execution if the worker binary is unavailable. Also fixes strict shutdown validation (ok is True, not truthiness) and replaces the false-positive real workspace test with the canonical minimal_ics55_pdk_factory fixture that requires success.
…fallback Wire a canonical workspace step-log resolver into the production worker route so that RunOperation archives EDA output to the correct step log paths (<workspace>/<step>_<tool>/log/<step>.log). Remove the binary-missing fallback to engine_flow.run_steps() — a missing worker binary now returns a structured OperationResult with error detail instead of silently falling back to in-process execution. Propagate OperationResult failure fields (error, exit_code, repaired_steps) into the CLI CommandResult error records for richer failure diagnostics.
…nt, and resilient drain LogStreamReader now accepts a valid_steps allowlist built from flow.json. Markers with (step, tool) pairs not in the set are treated as ordinary stderr data, preventing untrusted marker strings from switching archive ownership. After resolving a path, enforce that it resolves under workspace_dir before opening the archive file. This prevents path traversal via crafted marker step names like "../../escape". Resolver and on_output callback exceptions are now isolated: first error is recorded, the failed sink is disabled, and draining continues to EOF so pipe backpressure cannot deadlock the worker. The production CLI route reads flow.json to build the allowlist before starting RunOperation.
Emin017
force-pushed
the
emin/refactor-cli-worker
branch
from
August 12, 2026 01:40
85e5db4 to
b90817b
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What Changed
Scope
Select the areas touched by this PR:
uv.lock, or release artifacts.Runtime And Packaging Impact
ecc-toolsorecc-dreamplacedependency changedNotes:
Validation
List the commands you ran. Mark checks that are not applicable as N/A.
uv run pytest test/uv run ruff check chipcompiler testuv run ruff format --check chipcompiler testecc --help,ecc --version,ecc version --jsonnix run .#cli -- --helpSkipped checks and reason:
Checklist