fix(#195): stream opentofu task logs incrementally and expose log fields in the task list

[jgruberf5]

Summary

Two distinct defects (#195) made a running opentofu module opaque: logs_full_size stayed 0 for the whole run and the task list endpoint dropped log fields entirely.

Defect 1 — logs buffered until completion (root cause)

OpenTofuRuntime.run_{init,plan,apply,destroy} executed tofu via a blocking subprocess.run(capture_output=True), which returns the entire log only when the process exits. The Celery tasks accumulated that into all_logs and wrote task.logs (and therefore logs_full_size) exactly once, at the end. Result: 12 in-flight samples at 0, then the full size at once.

Fix: the run_* methods now take an optional on_output callback and, when supplied, stream stdout+stderr line-by-line through a new _stream_subprocess helper (Popen + a watchdog timer that preserves the (returncode, output) contract and TimeoutExpired semantics). When no callback is passed the classic blocking capture is kept, so existing behaviour and the runtime test suite are unchanged. A new TofuLogStreamer turns the callback into throttled task.logs writes, so logs_full_size grows during the run. The task still writes the complete all_logs at completion, so the final content, the final size, and the per-line timestamp format are all intact.

Defect 2 — list endpoint omitted log fields (root cause)

get_tasks never returned logs_full_size (it surfaced as null), while get_task did — so any client enumerating a module's tasks saw every task as empty.

Fix: the list now includes logs_full_size, computed in SQL via func.length(Task.logs) — logs is a deferred Text column, so this measures the size without loading every body — and matches the detail endpoint's len(logs). logs_truncated is included as False (the list returns no body by design; fetch the detail endpoint for logs). The frontend Task type already declared both fields optional, so the change is additive/backward-compatible.

What the tests lock

• test_opentofu_runtime_streaming.py: a real-subprocess handshake proving each line reaches on_output before the process exits (a buffered implementation would deadlock and time out); stderr-merge + exit code; timeout raises with partial output; the run_* methods route through streaming only when on_output is given, blocking capture otherwise.
• test_tofu_log_streamer.py: base+buffer composition, strict growth across lines, interval throttling, first-line force-flush, and commit-failure rollback (never raises).
• test_opentofu_tasks.py::TestOpenTofuTaskLogStreaming: task-level reproduce — the persisted logs_full_size (SQL length, exactly the detail computation) grows across lines while status is still in_progress; plus a guard that the plan task hands run_plan a callable sink.
• test_routes_tasks.py::TestTaskListLogFields: the list exposes logs_full_size matching the detail endpoint, and reports 0 (not null) for a task with no logs.

Mutation-checked: disabling the flush fails the streamer + task-level reproduce tests.

Closes #195

https://claude.ai/code/session_01UpRYiFserdBE5ESHn759N4

[jgruberf5]

Self-review (cold, adversarial) — production code sound; one vacuous test fixed

An independent cold auditor reviewed this PR, executing the code and mutation-testing every claim. No production blocker — the streaming, the throttled flush, the list-endpoint fields, and the contract preservation are all correct. One MAJOR test-vacuousness defect, now fixed.

MAJOR (fixed, 0906043) — the flagship streaming test proved nothing. test_delivers_each_line_before_process_finishes claimed a buffered impl "would deadlock and raise TimeoutExpired," but the child self-released after ~10s (range(200)*0.05) and the test asserted only line order — which a fully-buffered _stream_subprocess also satisfies. Reproduced: mutating _stream_subprocess to fire on_output only after proc.wait() (the exact #195 bug) left the test green.
→ Fixed: the child now blocks ~300s (>> the 8s watchdog), so a buffered impl can't self-release — it deadlocks, the watchdog kills it, and _stream_subprocess raises TimeoutExpired → the test errors. Added an explicit timing bound (< 5s) as a fast second signal. Mutation-confirmed: real code passes in ~5s; the buffered mutation now fails at the watchdog (~14s).

Held under attack (verified clean):

• Tail-drop under throttling — every call site appends the returned output and commits task.logs after the run, so final logs/logs_full_size is always complete; begin() forces an immediate first-line flush.
• func.length vs len() unicode — SQL LENGTH() is char-count (not bytes) on SQLite+Postgres; reproduced with accented/CJK/emoji strings — equal to Python len() every time; NULL→0 matches the detail endpoint.
• OpenAPI/TS staleness — GET /api/tasks has no response_model (raw dict, {} schema), so adding dict keys can't change the spec; --check up to date.
• Contract preservation — no-callback path byte-identical; watchdog is a real proc.kill(); TimeoutExpired.output still yields the partial log; deadlock/early-close/hung-child all terminate via the watchdog. (One intentional MINOR: stderr now interleaves with stdout chronologically — the downstream regex consumers are order-independent.)
• Lock safety — the streamer's mid-run db.commit() doesn't release the workspace lock (persistent lock-row + fence token + heartbeat, designed to survive commits).
• Mutation tests — list serializer + streamer flush both non-vacuous.

Full regression sweep: 82 passed (opentofu tasks/runtime, routes/tasks, + the two new files). Both #195 defects (buffered logs; list omits log fields) are correctly and now non-vacuously fixed.

[bonnyr-f5]

Review — round 1 @ 0906043

Reviewed under the review-discipline pipeline: invariant sweep across the whole surface, an independent cold full-diff audit with no session context, then every load-bearing claim re-verified locally before being written down here. CI gates run verbatim (results at the bottom).

Verdict: BLOCK — two blockers in the new _stream_subprocess. Defect 2 (the list endpoint) is solid and I'd take it as-is.

---

Defect 2 — list endpoint: verified correct

• func.length(Task.logs) ≡ len(logs) for every value the column can hold. NULL → SQL NULL → or 0, matching the detail endpoint's len(logs) if logs else 0. PostgreSQL length(text) counts characters, not octets, so it agrees with Python len() on multibyte input too.
• models/task.py:40 really is deferred(Column(Text)), so the comment's justification holds; add_columns adds a scalar expression and does not undefer the attribute.
• total = query.count() is computed before .add_columns(), so pagination totals are unchanged. Both joinedloads are many-to-one → no row fan-out and no subquery nesting under limit/offset.
• No response_model on the route, so the added keys are purely additive; frontend-v2/src/types/tasks.ts:34-35 already declares both optional; make openapi-check passes. The logs body itself is still absent from list rows.

---

Defect 1 — _stream_subprocess is a partial reimplementation of subprocess.run

It drops the two things subprocess.run does for correctness: closing the pipe, and killing the child on abrupt exit. The PR states it "preserves the (returncode, output) contract and TimeoutExpired semantics" — that does not hold, and both failures below were reproduced rather than reasoned about.

B-1 (blocker) · the timeout is not actually enforced — opentofu_runtime.py:88-98

The deadline is enforced only indirectly, via "kill the direct child ⇒ the pipe reaches EOF." That implication is false whenever any descendant inherited the write end of the pipe. Running the exact shape of the loop against a child that spawns a grandchild inheriting stdout:

PR _stream_subprocess: returned=False after 12.0s (deadline was 2s)
subprocess.run:        raised TimeoutExpired after 2.0s

proc.stdout is never closed on any path, and proc.wait() at :96 has no timeout, so nothing bounds the call. At APPLY_TIMEOUT the watchdog SIGKILLs tofu, the read loop does not return, TimeoutExpired is never raised, and the task never completes — while the project_modules lease is never released and its heartbeat thread keeps refreshing it, so the stale-lock sweeper cannot reclaim it either. The module ends up permanently locked.

On reachability, being exact because it cuts the other way from how it first looks: grepping for local-exec / null_resource / data "external" returns zero hits — but the repo ships zero .tf files at all. ModuleLibrary.git_source (backend/models/module.py:34) is nullable=False: every module tofu executes is fetched from external git. So that grep establishes nothing about production, and the HCL is operator-supplied.

Fix shape: close proc.stdout in a finally and proc.wait(timeout=…), or use with proc: and read against a deadline.

B-2 (blocker) · non-timeout exceptions orphan a live tofu apply — opentofu_runtime.py:88-98

The try carries only finally: timer.cancel() — no proc.kill(), no pipe close. CPython's subprocess.run has except: # Including KeyboardInterrupt → process.kill(); raise, plus with Popen(...) closing the pipes on exit.

Celery delivers SoftTimeLimitExceeded by raising it in the worker's main thread — which, during a streamed run, is blocked inside for line in proc.stdout. Handlers for it already exist at opentofu_tasks.py:943 (apply) and :1216 (destroy). Before this PR that path was safe precisely because subprocess.run killed the child. Now tofu survives the worker, keeps mutating cloud state and the .tfstate under work_dir while the task is marked failed and the workspace lock is released — and a retry then runs a second concurrent tofu against the same state directory.

on_output exceptions specifically are guarded (:93-95), so the sink cannot trigger this; everything else can (soft/hard time limit, UnicodeDecodeError from text=True on non-UTF-8 provider output, OSError on the pipe).

---

Minor

F1 · the persisted log rewinds on the stale-plan retry path — opentofu_tasks.py:769, 788, 808

all_logs += apply_logs is at :811, but streamer.begin(all_logs) is called at :769, :788 and :808 — all inside the retry block, all before :811. The first apply's output has already been streamed and persisted; the retry's base omits it, so task.logs / logs_full_size shrink mid-run, contradicting the PR's own strict-growth invariant.

Swept all 12 begin() sites: the plan task (:449, :471), _ensure_workspace_initialized (:633), the main plan (:724) and all three destroy sites (:1084, :1106, :1124) are correctly caught up — the class is confined to those 3 retry sites. The component run reports opentofu_tasks.py:766-784, 792-803 uncovered, which is exactly why no test sees it.

Fix shape: move the append to immediately after the first apply (which also recovers the first apply's output, currently dropped — pre-existing), or have begin() reject a base shorter than what it has already persisted.

M-1 · a comment that is factually wrong, in the one unguarded branch — opentofu_runtime.py:1991

Measured locally: subprocess.run(text=True, timeout=…) populates TimeoutExpired.stdout as bytes (b'x\n'), stderr as None. Five timeout branches carry isinstance(stdout, bytes) decode guards (:1664, 1715, 1768, 1834, 1926) — the codebase already knows this. run_destroy instead asserts the opposite in a comment and has no guard, so it raises TypeError: can't concat str to bytes on the blocking path.

Pre-existing — but this PR makes run_opentofu_destroy always pass a sink, so that caller silently starts working while opentofu_engine.py:452 still crashes. Same method, two behaviours, decided by an unrelated kwarg.

M-3 · write amplification — _tofu_helpers.py:77-86

The throttle is time-only (interval=2.0) and every flush rewrites the entire accumulated log. A 90-minute apply is ≈2700 full UPDATE tasks SET logs = <N bytes>; total payload ≈ 1350·N. On PostgreSQL logs is TOASTed above ~2KB, so each UPDATE rewrites the whole TOAST chain — thousands of dead heap + TOAST tuples per task. Wants a size-aware trigger, an append (logs = logs || :chunk), or a buffer cap.

M-5 · _flush's except performs a session-wide rollback — _tofu_helpers.py:81-86

self._db.rollback() is not scoped to the log write: it discards everything pending in the task's session and expires the whole identity map, silently, with no log line. I traced what is actually dirty at every sink point in all four task functions and found no currently exploitable instance — task.command is the realistic casualty, and WorkspaceManager.mark_initialized/clear_plan both self-commit. So this is a live class with no live instance: the next pre-run mutation anyone adds inherits it. Also self._last = now is set before _flush(), so a failing flush still consumes the throttle window.

routes/tasks.py:97-105 — the "avoids loading every task's full log body" claim holds on the wire but not server-side. PostgreSQL's textlen() fast-paths to toast_raw_datum_size() only when the database encoding is single-byte; on UTF-8 it must detoast and run pg_mbstrlen_with_len, i.e. fetch and decompress every log body in the page. With limit up to 200 on a 60s poll, that is the exact cost deferred() exists to avoid. Worth softening the comment even if the cost is acceptable.

Test wiring — only run_opentofu_plan's sink is asserted. Deleting on_output= from apply or destroy leaves the whole suite green. test_timeout_kills_and_raises_with_partial_output never asserts the process actually died.

Nits

• :72 — bufsize=1 is cosmetic; line_buffering on a TextIOWrapper affects writes. Streaming works because readline() returns at the first newline regardless.
• :89 — assert proc.stdout is not None vanishes under python -O. No PYTHONOPTIMIZE in any image here, so currently harmless.
• :84 — _kill_on_timeout sets timed_out without checking proc.poll(), so a process finishing microseconds before the deadline still reports a timeout and discards a successful returncode plus the full output.

---

Falsified hypotheses (so coverage is distinguishable from silence)

Two things that look like they should be bugs and are not:

• The mid-run db.commit() does not release the workspace lock. module_lock is a fence-token row lease with a heartbeat (entity_lock.py), not pg_advisory_xact_lock and not SELECT … FOR UPDATE; a commit cannot drop it.
• It does not race the heartbeat writer. HeartbeatRefresher uses its own SessionLocal() (entity_lock.py:442), so the streamer's commits never share a Session across threads. The sink genuinely runs on the task's own thread, as the docstring claims.

Gate results — CI targets run verbatim @ 0906043

gate	result
make lint-backend (ruff)	PASS
make openapi-check	PASS — spec up to date (533 paths / 472 schemas)
make test-backend-unit	PASS — 4862 passed
make test-backend-component	PASS — 3197 passed
make test-integration	PASS — 760 passed, 207 deselected
make typecheck-backend (mypy)	not run — the gate covers only core/ schemas/; no changed file in this diff is in scope

Two honest gaps in this review: mypy covers none of the changed files, and the func.length equivalence is only ever exercised on SQLite (conftest.py pins in-memory SQLite) — never on the production dialect where TOAST and encoding-dependent textlen() live.