refactor: isolate worker stdio and stream tool logs - #210
Draft
Emin017 wants to merge 55 commits into
Draft
Conversation
Emin017
force-pushed
the
emin/refactor-cli-worker
branch
from
August 12, 2026 01:40
85e5db4 to
b90817b
Compare
Emin017
marked this pull request as ready for review
August 12, 2026 03:11
5 tasks
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 18, 2026 07:09
54055d4 to
a1e53f1
Compare
…ites Add a protocol version field (v: 1) to step marker payloads; parsers reject frames with a missing or unsupported version as ordinary bytes. Relocate the end marker in EngineFlow.run_step so it fires after all step-scoped writes (final state persistence, [RESULT], QOR, layout snapshot, db cleanup) and before the completion observer notification, so a consumer that has read the end marker has seen every byte of the step.
Normative specification of the step marker byte-stream protocol: frame format and v1 payload, consumer/producer semantics, the ordering guarantee (end after all step-scoped writes, before completion notify), the single-producer invariant, the archive path layout, and the DEC-1 protocol change making the GUI the only live-log consumer.
…alls to RunOperation FlowRunStepRequest gains an optional reset_dependents field (additive, mirroring operation.start_step), and the flow.run_step handler forwards it so direct step reruns can invalidate the downstream suffix. RunOperation.run_sequence executes an ordered list of RPC calls in one worker session, stopping at the first failed RPC: remaining calls are skipped, the session is still shut down gracefully and drained, and the returned OperationResult describes the failing call.
All execution paths now go through RunOperation; the TTY distinction only selects UI rendering. - run --workspace maps --resume/--from onto flow.run_step with reset_dependents plus a follow-up flow.run in one worker session, and --only/--force onto a single flow.run_step; the no-op cases (already successful selections) keep their current records. - run_flow_with_progress is rewritten around the reader callbacks: begin markers drive step transitions, on_output drives the throttled live line, and per-step final states refresh from flow.json on each begin marker and once at operation end. - LogStreamReader gains an on_step_event callback fired on matched begin/end markers; RunOperation forwards it. - Delete preserve_cli_stdio, the log monitor, the incremental log tail, redirect_stdio_to_file, and the in-process rerun execution.
Executors never write step log files, so there is nothing for the runtime server to tail: remove the step log tail thread, the delta publisher, the final log reader, and their call sites. step.completed keeps step, tool, state, stepCommitId, workspaceRevision, and the render gate; live step.log events and finalLog are now synthesized by the archiving client instead of the executor.
parse_marker already catches UnicodeDecodeError, but the rejection was not pinned by a test. Add the normative case so the Python reader and the TS archiver (fatal TextDecoder) are held to the same rule: a frame whose payload is not valid UTF-8 is ordinary stream bytes, never a marker.
Cover the resume/suffix dependency path the filter exists for: with executable_steps active, a non-executing Success predecessor is built without a dependency check, so its outputs still chain into the executing successor's inputs and its missing tool marks nothing Incomplete.
Round-5 re-review P3 follow-ups: pin rejection of non-object JSON payloads ([], null, 42, "hello", true) in the normative parse matrix, and give the end-marker ordering step a feature path so the QOR/metrics refresh is explicitly ordered before the end marker.
set_state returned True even when its save failed, and run_step's redundant second save only flipped a local variable — a failed final save left the canonical record (and possibly flow.json) at Success while reporting Imcomplete. set_state now returns the real save result, run_step performs one final save, and on failure the canonical record is downgraded in memory, the end marker suppressed, and Imcomplete reported. The regression uses a real Flow fixture so the failing save is the one persisting the record: exact save count, no end marker, Imcomplete return/observer/record, Ongoing on disk.
The invalidate_dependents path persisted and deleted the target through _prepare_steps_for_rerun, then saved downstream records separately — a second-save failure left target artifacts deleted and the session records half-mutated. _prepare_steps_for_rerun now validates paths, snapshots every affected record, applies target reset plus downstream state invalidation, persists once, and restores the snapshots before raising on failure; artifacts are cleared only after the save succeeds. The now-unused _invalidate_step_records is removed. Regressions: a three-step save failure pins restored records, untouched artifacts, and a single save attempt; an in-process flow_run_step run with a real EngineFlow proves a failed final save leaves no Success record and the next non-rerun call re-executes the step.
Deleting redirect_stdio_to_file broke agent/engine.py at import time (main pytest collects only test/, so the break was invisible), and its run_step still redirected executor stdio into step log files with no markers — against the protocol. The agent now emits begin/end markers exactly like EngineFlow.run_step, never touches step log files, and its _finish_step returns the authoritative save result: a failed final save downgrades the canonical record and suppresses the end marker. New tests pin the marker ordering around step writes and the failed-save suppression. The three-step invalidation regression now also seeds non-default record metadata and asserts identity-preserving rollback of the complete records.
AgentEngineFlow duplicated the full step lifecycle — selection, Ongoing transition, marker emission, memory tracking, tool invocation, final save, downgrade, post-processing, db cleanup — which is how the redirect_stdio_to_file deletion left it broken and protocol-inconsistent for four rounds. EngineFlow.run_step now invokes the tool through _invoke_step_tool and derives the state through _derive_step_state; the base hooks preserve existing behavior exactly. AgentEngineFlow keeps only its DRC insertion and two hook overrides (run_agent_step, with False -> Imcomplete, Invalid passthrough, True|Success -> artifact check). New agent regression proves the inherited lifecycle drives the agent hook and opens no step log file even when one is declared.
Emin017
marked this pull request as draft
August 19, 2026 07:32
The run_step+flow.run sequence let the trailing unscoped flow.run resume from the first non-success step — with a failed step before the --from boundary, it executed and mutated steps outside the requested suffix. The suffix is now driven as explicit per-step run_step calls (boundary step with reset_dependents, then each persisted successor), preserving rerun.run_from's exact scoping and stop-on-first-failure ordering. Wiring tests updated to the stepwise contract, plus a regression with a failed step before the --from boundary proving it is never called.
Direct EngineFlow execution (agent candidate reruns, documented examples) emitted markers to fd 2 with no client to consume them: raw frames leaked to the caller's stderr and no per-step archive was written. The new archive_own_step_logs context redirects the process's own fd 2 through a pipe so a LogStreamReader archives step bytes and consumes markers while echoing everything to the original stderr — the executor still never opens a log file; the client role just runs in-process. The agent's candidate rerun path uses it, and the teardown closes the saved stderr only after the reader drains, pinned by a capfd regression.
archive_own_step_logs redirected only fd 2, but tool subprocess stdout and native tool logging write fd 1 — in-process runs (agent candidate reruns, direct EngineFlow examples) lost those bytes from the step archives. Both descriptors now share the pipe, matching the merged stream the CLI worker's stdio isolation produces; markers stay on fd 2 and every byte echoes to the original stderr. The regression now also pins fd 1 bytes landing in the archive in write order.
The public rerun helpers (run_from/run_only/run_resume) execute steps in-process with no client to consume the marker stream: step logs stayed empty and raw ECC-STEP frames leaked to the caller's terminal. _run_selected now wraps execution in archive_own_step_logs so direct callers get the same client-side archival as worker runs. A regression drives real marker+byte writes through run_from and asserts the archive contents and a marker-free terminal.
…ir failures Two worker failure paths left persisted state inconsistent with the reported result. When the worker completed but the reader archived with errors (or an unmatched begin), the step's Success record survived while the CLI reported failure — resume would skip the step and never recreate the missing log; the error branch now repairs the affected record to Incomplete. And a crash-repair that cannot persist (disk full, permissions) was silently suppressed; both error paths now append the repair failure to the result error instead. Regressions pin the Success downgrade on archive failure and the surfaced repair failure on crash.
…e failures - step_log_archive_resolver mirrors the sizer builder's sanitized directory (Timing optimization -> timing_optimization_sizer), so worker success no longer writes to a directory the built step never owned. - Every remaining direct EngineFlow entry point (integration conftest, both gcd examples) now runs inside archive_own_step_logs; the conftest's obsolete fd save/restore workaround is removed. - The reader records error_step at the first archive error, and all three worker failure paths (clean-shutdown errors, RPC error, crash) reconcile through one _reconcile_step_state helper — a Success record never survives a missing or incomplete archive. - In-process rerun paths (engine.rerun helpers, agent candidate reruns) now fail and downgrade the record when archival fails or a begin is unmatched, instead of reporting success over a missing log. - Rerun preparation moves to runtime/rerun_prepare.py, keeping workspace_api.py an adapter per the module-size rule.
…ample - LogStreamReader drained pipes with read(8192), which blocks until the buffer fills or EOF: steps emitting less than 8 KiB showed no live progress until they exited. read1 delivers whatever the pipe currently holds (with a plain-read fallback); a pipe-based regression proves a short line is delivered while the writer stays open. - docs/examples/gcd/ics55flow.py was the last shipped entry point calling run_steps() bare; it now runs inside archive_own_step_logs like the other direct-run examples.
…pipe - _run_selected reconciles the reader state in an exception path too: a step raising after its begin marker no longer skips the archive downgrade before the exception propagates. - Both downgrade sites now go through set_state, so the repair persists through the authoritative save and a failed save is surfaced (logged) instead of leaving flow.json at Success over a missing archive. - archive_own_step_logs closes the pipe read stream at teardown; long runs of in-process reruns no longer accumulate pipe fds toward EMFILE. A regression proves a post-begin exception still downgrades the record.
- prepare_steps_for_rerun restores the persisted records when post-save cleanup (artifact delete, subflow/checklist reset) fails midway, so the workspace is never left with Unstart states over half-deleted outputs. - EngineFlow.run_step downgrades the persisted Ongoing when the begin marker cannot reach fd 2 — no reader ever saw the step, so worker recovery could never identify it. - Runtime flow_run_step appends the global flow/status log after each executed step, restoring the log_flow side effect the in-process rerun helpers used to provide for --workspace/--from/--resume runs. Regressions pin the cleanup rollback and the begin-marker downgrade.
Review rounds kept finding the same gap: direct EngineFlow execution is a documented Python API, but per-callsite wrapping cannot hold. run_steps now wraps itself in archive_own_step_logs, so bare documented scripts archive step logs and keep markers off the terminal. The context passes through untouched when an outer client owns the stream (the stdio server entry marks itself via mark_external_log_client) or when nested inside another archive context, preserving the single-producer invariant for worker, sidecar, agent-candidate, and rerun-helper paths. The integration conftest drops its now-redundant explicit wrapper. Regressions pin the nested and external-client passthroughs.
… and pre-marker crashes - prepare_steps_for_rerun cleanup is per-step: a mid-cleanup failure keeps the persisted Unstart on steps whose artifacts are already gone and rolls back only untouched steps, so resume never trusts Success over deleted outputs. - run_steps reconciles the self-archive reader after the loop (and before exception propagation): archive failures or unmatched begins downgrade the record and return False instead of reporting success over a missing log. - Worker crash recovery falls back to the persisted Ongoing record when no stream evidence exists (kill between the Ongoing save and the begin marker). Regressions pin all three paths.
- on_output/on_step_event callback exceptions now land in LogStreamState.display_error instead of error: a broken renderer no longer reads as an archive failure and can no longer downgrade a step whose archive is complete. - downgrade_unarchived_step surfaces an unpersistable downgrade with an explicit error log naming the stale Success record, instead of reporting a repair that never reached disk. Regressions pin both: a crashed renderer leaves the operation successful with the record intact, and a failed downgrade save keeps the disk record honestly at Success while the operation fails.
…istence The two callback-resilience contract tests now assert display_error (display-only failures), and the rerun suite gains a regression proving an unpersistable downgrade keeps the disk record honestly at Success while the operation reports failure.
…, reset the self-archive guard - The agent candidate path reconciles reader evidence before propagating a step exception, matching rerun/run_steps: no persisted Success over a partial archive. - RunOperation converts parent SIGTERM into KeyboardInterrupt on the main thread (restored on exit), routing through crash recovery so the worker's process group and EDA descendants are reaped instead of mutating the workspace after the CLI dies. - archive_own_step_logs setup failure now restores any redirected fds, closes opened descriptors, and resets the guard — a failed setup no longer poisons later in-process runs. Regressions pin all three paths.
…suffix tools - EngineFlow.run_step now also self-archives (passthrough inside worker/ sidecar processes and nested contexts), so no direct execution path leaks markers or leaves step logs unwritten. - repair_flow_state and downgrade_unarchived_step match on (name, tool): a duplicate step name under another tool is no longer downgraded by an unrelated archive failure; the reader carries error_tool for the evidence pair. - The CLI preflights every selected step's tool before the first worker call, so an unavailable tool fails the run before reset_dependents invalidates and clears the suffix. Regressions pin the pair matching, the preflight, and the direct-run archival.
… the 700-line bar - engine/runner.py: EngineFlowRunner mixin owns the step/flow execution lifecycle (markers, memory tracking, authoritative final save, post-processing, db cleanup, observer/render gate); flow.py keeps the data/state/build spine at 321 lines. run_step now reconciles the self-archive reader on both success and exception paths. - cli/command_handlers/workspace_run.py: worker-call construction, suffix preflight, and outcome reconciliation move out of project.py (499 lines). The preflight now catches dependency-check exceptions and covers the sizer runtime sentinel (src/sizer_os.tcl), and archive_own_step_logs guards descriptor-duplication failures so a broken setup releases the guard and restores fds. Regressions pin direct run_step downgrade, preflight rejection without mutation, and setup-failure recovery.
…den archive teardown - A begin rejected by name/containment validation now keeps the marker's (step, tool) identity on error_step/error_tool, so failure-path reconciliation can find and downgrade the step even though it never activated. - archive_own_step_logs moves the pre-setup flushes into the guarded try and releases _SELF_ARCHIVE_ACTIVE in an innermost finally with per-step suppression, so any teardown failure (broken descriptor, closed stream) still restores fd 1/2 and frees the guard. Regressions pin the identity preservation and teardown failure recovery.
…flight the worker, split worker-op tests - WorkerClient.request reads error.data.message (RuntimeServer's channel for RuntimeApiError's actionable text) with a fallback to the top-level message, so failures name the failed step instead of a bare command_failed. - archive_own_step_logs teardown joins the reader a second time after the stop signal and records a TimeoutError on the reader state when the drain never completes, instead of releasing the guard over a live reader. - preflights worker availability before creating the run directory (and before --overwrite deletes an existing one), so a missing worker leaves the workspace untouched for a clean retry; the orphaned duplicate of the check in project.py is gone. - test_worker_operation.py splits into lifecycle/sequencing, crash recovery, and archive modules over a shared worker_operation_support preamble, each below the 700-line review bar. Regressions pin the nested-message parsing, the preflight (run dir uncreated; overwrite target preserved), and all split suites pass.
…r host, track completed calls - Database initialization moves inside the step's marked lifecycle (immediately after the begin marker) and the caller-side pre-inits in run_steps, rerun._run_selected, runtime _flow_run_step, and the agent candidate path are gone: native initialization output now reaches the step archive instead of leaking unscoped, per the marker-protocol ordering rule the spec now states explicitly. - EngineFlowRunner declares its host contract (workspace, workspace_steps, engine_db, and the record helpers) with class-level annotations and TYPE_CHECKING stubs, so type checkers no longer see 45 unknown-member errors in the extracted lifecycle. - RunOperation counts completed worker calls on OperationResult, and the workspace-run outcome derives executed/failed steps from that count: a worker dying during startup no longer reports already-successful selected steps (--from / --only --force) as executed from stale flow.json states. - The worker preflight rejects non-executable binaries (os.access X_OK) before --overwrite deletes an existing run. Regressions pin begin-before-init ordering, the startup-failure outcome, the executable-bit rejection, and the sizer engine-clearing semantics under the per-step init.
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
2 participants
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.
Completes the executor log-streaming refactor: executors never write step
log files — they write bytes to fd 1/2 plus versioned step markers on fd 2,
and each client archives per-step logs. Includes the v1 marker protocol
(docs/specification/marker-protocol.md), the end-marker ordering fix, full
CLI unification on the worker (all
ecc runvariants), and the deletion ofthe server-side step-log tail machinery.
head without the accompanying GUI PR breaks GUI live step logs (the GUI's
Electron archiver is what restores them). The superproject pin bump lands
in the GUI PR, atomically with the archiver.
Validation
pytest test/ --ignore=test/integration: 1522 passed, 9 skipped,13 xfailed;
pytest agent/test/: 129 passed; ruff lint+format clean.clean recursive checkout of its head: focused ecc suites 79 passed,
agent 129 passed, GUI archiver/bridge suites 79 passed, all lint/type
gates clean).