refactor: isolate worker stdio and stream tool logs - #210
Draft
Emin017 wants to merge 35 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.
…get checks Review follow-ups for the worker-streamed logs delta: - emit_step_marker now drains C/C++ stdio buffers (flush_cstdio) before writing, so native output can never land on the wrong side of a step boundary. - parse_marker rejects JSON boolean versions (true == 1 in Python). - LogStreamReader validates step/tool names (no separators, dot segments, or empty values) and containment before activating a step; violating begin frames degrade to ordinary bytes instead of being consumed. - ecc run --workspace --only maps executed steps to flow.run_step with rerun: true, restoring the previous clean-artifact contract; --force remains the gate for re-executing a successful step. - Add CLI-level tests that execute against a real worker subprocess: --resume suffix execution with archiving, --only single-step execution, and non-TTY flow.run archival with no marker leakage.
The documented workspace-mode contract requires a re-executed step to replace its own artifacts and mark downstream steps Unstart while keeping their outputs. reset_dependents cannot express this because it also clears downstream artifacts, so FlowRunStepRequest gains an additive invalidate_dependents field: the server prepares only the target step and marks the downstream suffix Unstart in flow.json. Also resolve the remaining review findings: the log reader now resolves and validates an archive target exactly once and opens that validated path, and the C stdio ordering test uses an explicitly buffered FILE* so it fails without the flush.
- The log reader processes lines with a single split per chunk, trims its tail only past twice the cap, shares path_is_within for containment, and records first-error-wins through one helper. - RunOperation's failed-RPC branch reuses _handle_protocol_or_crash. - The CLI shares one worker-call entry point and one tolerant flow.json reader (cli.inspection.discovery.read_flow_json); the TTY live line throttles before decoding and sanitizing. - The rerun preparation collapses the duplicate prepare branches. - The marker protocol spec now documents the allowlist policy and each consumer's failure default.
The producer never inserts a newline before a marker frame, so a frame can immediately follow tool output that lacks a trailing newline. The reader now scans for the reserved prefix anywhere in the stream: bytes before a candidate are data, a trailing partial prefix is held back, incomplete candidates are bounded, and only valid newline-terminated v1 frames are consumed. The specification documents the scanning rule and now shows the exact wire spelling of the prefix.
A flow that raises after the begin marker returns a structured RPC error from a still-alive worker, leaving flow.json with a stale Ongoing record. The RPC-error path now performs the same repair as crash recovery after the reader drains: the active step is marked Incomplete and the result carries repaired_steps alongside the failing RPC response.
test_log_stream.py crossed the repository's 700-line review bar at 814 lines. The coverage now lives in three focused modules, every assertion preserved: test_log_stream.py keeps the reader archiving and resilience cases, test_log_stream_markers.py holds the parse/emit/boundary-scanning suite (including the chunked-stream helper), and test_log_stream_targets.py covers allowlist, sanitization, containment, resolver, and step-event behavior.
- run_step no longer emits the end marker when the final state could not be persisted; the step downgrades to Imcomplete so consumers see a crashed step rather than a stale flow.json claiming success. - repair_flow_state now also repairs a Success record for the step that was active at crash time: a Success without a completed end marker means the crash interrupted post-processing, so the persisted result is not trustworthy. - Rerun invalidation is atomic again: the invalidation is persisted before any artifact directory is deleted, and a failed save refuses to modify outputs (matching the previous CLI contract). - The worker's flow build verifies tool dependencies only for steps that will actually execute, so resume/from/only runs reuse successful predecessor outputs when an unselected tool is absent. - The 512-byte marker holdback is now inclusive per the specification; exactly-512-byte incomplete candidates are held and consumed.
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
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).