diff --git a/backend/routes/tasks.py b/backend/routes/tasks.py
index e3acf0b..ab11774 100644
--- a/backend/routes/tasks.py
+++ b/backend/routes/tasks.py
@@ -13,7 +13,7 @@
 
 from fastapi import APIRouter, Depends, Query, WebSocket, WebSocketDisconnect
 from pydantic import BaseModel
-from sqlalchemy import desc
+from sqlalchemy import desc, func
 from sqlalchemy.orm import Session, joinedload
 
 
@@ -92,14 +92,22 @@ def get_tasks(
     # Order by creation time (newest first) and paginate
     # Eager load relationships to avoid N+1 queries
     # B3: load_only() — task list only needs project.name and library_module.name
-    tasks = query.options(
+    #
+    # #195: expose the log size on the list too. ``logs`` is a deferred Text
+    # column, so we compute its length in SQL (func.length) rather than loading
+    # every task's full log body just to measure it — this keeps the list cheap
+    # while giving clients the same ``logs_full_size`` the detail endpoint
+    # reports. The full ``logs`` body stays detail-only by design.
+    rows = query.options(
         joinedload(Task.project).load_only(Project.id, Project.name),
         joinedload(Task.module).joinedload(ProjectModule.library_module).load_only(ModuleLibrary.id, ModuleLibrary.name)
+    ).add_columns(
+        func.length(Task.logs).label("logs_full_size")
     ).order_by(desc(Task.created_at)).limit(limit).offset(offset).all()
 
     # Format response
     tasks_data = []
-    for task in tasks:
+    for task, logs_full_size in rows:
         task_dict = {
             "id": task.id,
             "celery_task_id": task.celery_task_id,
@@ -118,6 +126,12 @@ def get_tasks(
             "exit_code": task.exit_code,
             "error": task.error,
             "archived": task.archived,
+            # NULL logs → length is NULL → report 0, matching the detail
+            # endpoint's ``len(logs) if logs else 0``. ``logs_truncated`` is
+            # always False here: the list never returns a truncated body (it
+            # returns no body at all — fetch the detail endpoint for logs).
+            "logs_full_size": logs_full_size or 0,
+            "logs_truncated": False,
         }
         tasks_data.append(task_dict)
 
diff --git a/backend/services/execution/opentofu_runtime.py b/backend/services/execution/opentofu_runtime.py
index 904fb63..fb0e0c0 100644
--- a/backend/services/execution/opentofu_runtime.py
+++ b/backend/services/execution/opentofu_runtime.py
@@ -23,7 +23,9 @@
 import shutil
 import subprocess
 import tempfile
+import threading
 import time
+from collections.abc import Callable
 from pathlib import Path
 
 from sqlalchemy.orm import Session
@@ -36,6 +38,72 @@
 logger = logging.getLogger(__name__)
 
 
+def _stream_subprocess(
+    cmd: list[str],
+    *,
+    cwd: str,
+    env: dict,
+    timeout: int,
+    on_output: Callable[[str], None],
+) -> tuple[int, str]:
+    """Run ``cmd`` streaming stdout+stderr line-by-line to ``on_output``.
+
+    Behaves like ``subprocess.run(..., capture_output=True, text=True,
+    timeout=...)`` from the caller's point of view: it returns
+    ``(returncode, combined_output)`` and raises ``subprocess.TimeoutExpired``
+    (with ``output`` populated) on timeout, so the existing timeout-handling
+    branches in each ``run_*`` method work unchanged.
+
+    The difference — and the whole point (issue #195) — is that output is
+    delivered incrementally: each line is handed to ``on_output`` the moment
+    OpenTofu emits it, letting the caller persist progress (``task.logs`` /
+    ``logs_full_size``) *during* a long run instead of only at completion.
+    stderr is merged into stdout (``STDOUT``) so lines interleave in the order
+    they were produced. A watchdog timer kills the process at ``timeout``,
+    mirroring ``subprocess.run``'s timeout semantics.
+    """
+    proc = subprocess.Popen(
+        cmd,
+        cwd=cwd,
+        env=env,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        bufsize=1,  # line-buffered so lines surface as they are produced
+    )
+
+    timed_out = threading.Event()
+
+    def _kill_on_timeout() -> None:
+        timed_out.set()
+        try:
+            proc.kill()
+        except Exception:  # noqa: BLE001 — process may have already exited
+            pass
+
+    timer = threading.Timer(timeout, _kill_on_timeout)
+    timer.start()
+
+    chunks: list[str] = []
+    try:
+        assert proc.stdout is not None
+        for line in proc.stdout:
+            chunks.append(line)
+            try:
+                on_output(line.rstrip("\n"))
+            except Exception:  # noqa: BLE001 — a log sink must never break the run
+                logger.exception("on_output callback raised while streaming subprocess output")
+        proc.wait()
+    finally:
+        timer.cancel()
+
+    output = "".join(chunks)
+    if timed_out.is_set():
+        # Match subprocess.run: surface a TimeoutExpired carrying what we read.
+        raise subprocess.TimeoutExpired(cmd, timeout, output=output)
+    return proc.returncode, output
+
+
 def _add_provider_lock_timeout_hint(output: str) -> str:
     """Annotate known provider install stalls with a concrete runtime hint."""
     normalized_output = (output or "").lower()
@@ -1516,7 +1584,43 @@ def write_provider_config(self, work_dir: str, module: ProjectModule, variables:
     APPLY_TIMEOUT = 90 * 60     # 90 minutes - long-running resource creation
     REFRESH_TIMEOUT = 30 * 60   # 30 minutes - state refresh
 
-    def run_init(self, work_dir: str, env: dict, timeout: int | None = None) -> tuple[int, str]:
+    @staticmethod
+    def _run_tofu(
+        cmd: list[str],
+        work_dir: str,
+        tofu_env: dict,
+        timeout: int,
+        on_output: Callable[[str], None] | None,
+    ) -> tuple[int, str]:
+        """Execute a tofu command, returning ``(returncode, stdout+stderr)``.
+
+        When ``on_output`` is provided the output is streamed line-by-line via
+        :func:`_stream_subprocess` (issue #195); otherwise it falls back to the
+        classic blocking ``subprocess.run`` capture. Both paths raise
+        ``subprocess.TimeoutExpired`` on timeout so each caller's existing
+        timeout branch is unchanged.
+        """
+        if on_output is not None:
+            return _stream_subprocess(
+                cmd, cwd=work_dir, env=tofu_env, timeout=timeout, on_output=on_output,
+            )
+        result = subprocess.run(
+            cmd,
+            cwd=work_dir,
+            env=tofu_env,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+        return result.returncode, result.stdout + result.stderr
+
+    def run_init(
+        self,
+        work_dir: str,
+        env: dict,
+        timeout: int | None = None,
+        on_output: Callable[[str], None] | None = None,
+    ) -> tuple[int, str]:
         """
         Run tofu init.
 
@@ -1524,6 +1628,9 @@ def run_init(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl
             work_dir: Workspace directory
             env: Environment variables
             timeout: Optional timeout in seconds (default: INIT_TIMEOUT)
+            on_output: Optional line callback. When provided, output is streamed
+                line-by-line (issue #195) so callers can persist progress during
+                the run; when None, behaviour is the classic blocking capture.
 
         Returns:
             Tuple of (exit_code, output)
@@ -1535,19 +1642,15 @@ def run_init(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl
         logger.info(f"Running tofu init in {work_dir} (timeout: {timeout}s)")
 
         try:
-            result = subprocess.run(
+            returncode, raw_output = self._run_tofu(
                 ["tofu", "init", "-no-color", "-input=false"],
-                cwd=work_dir,
-                env=tofu_env,
-                capture_output=True,
-                text=True,
-                timeout=timeout
+                work_dir, tofu_env, timeout, on_output,
             )
 
-            output = _add_provider_lock_timeout_hint(result.stdout + result.stderr)
-            logger.info(f"tofu init completed with exit code {result.returncode}")
+            output = _add_provider_lock_timeout_hint(raw_output)
+            logger.info(f"tofu init completed with exit code {returncode}")
 
-            return result.returncode, output
+            return returncode, output
 
         except subprocess.TimeoutExpired as e:
             # S14-021: Handle timeout for init
@@ -1565,7 +1668,13 @@ def run_init(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl
             output = _add_provider_lock_timeout_hint((stdout or "") + (stderr or "") + timeout_msg)
             return 1, output
 
-    def run_plan(self, work_dir: str, env: dict, timeout: int | None = None) -> tuple[int, str]:
+    def run_plan(
+        self,
+        work_dir: str,
+        env: dict,
+        timeout: int | None = None,
+        on_output: Callable[[str], None] | None = None,
+    ) -> tuple[int, str]:
         """
         Run tofu plan.
 
@@ -1573,6 +1682,7 @@ def run_plan(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl
             work_dir: Workspace directory
             env: Environment variables
             timeout: Optional timeout in seconds (default: PLAN_TIMEOUT)
+            on_output: Optional line callback for incremental log streaming (#195).
 
         Returns:
             Tuple of (exit_code, output)
@@ -1584,19 +1694,14 @@ def run_plan(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl
         logger.info(f"Running tofu plan in {work_dir} (timeout: {timeout}s)")
 
         try:
-            result = subprocess.run(
+            returncode, output = self._run_tofu(
                 ["tofu", "plan", "-no-color", "-input=false", "-out=plan.out"],
-                cwd=work_dir,
-                env=tofu_env,
-                capture_output=True,
-                text=True,
-                timeout=timeout
+                work_dir, tofu_env, timeout, on_output,
             )
 
-            output = result.stdout + result.stderr
-            logger.info(f"tofu plan completed with exit code {result.returncode}")
+            logger.info(f"tofu plan completed with exit code {returncode}")
 
-            return result.returncode, output
+            return returncode, output
 
         except subprocess.TimeoutExpired as e:
             # S14-021: Handle timeout for plan
@@ -1673,6 +1778,7 @@ def run_apply(
         env: dict,
         timeout: int | None = None,
         module: ProjectModule | None = None,
+        on_output: Callable[[str], None] | None = None,
     ) -> tuple[int, str, dict]:
         """
         Run tofu apply and capture outputs.
@@ -1681,6 +1787,7 @@ def run_apply(
             work_dir: Workspace directory
             env: Environment variables
             timeout: Optional timeout in seconds (default: APPLY_TIMEOUT)
+            on_output: Optional line callback for incremental log streaming (#195).
 
         Returns:
             Tuple of (exit_code, output, captured_outputs)
@@ -1693,21 +1800,16 @@ def run_apply(
 
         try:
             # Apply the plan
-            result = subprocess.run(
+            returncode, output = self._run_tofu(
                 ["tofu", "apply", "-no-color", "-input=false", "-auto-approve", "plan.out"],
-                cwd=work_dir,
-                env=tofu_env,
-                capture_output=True,
-                text=True,
-                timeout=timeout
+                work_dir, tofu_env, timeout, on_output,
             )
 
-            output = result.stdout + result.stderr
-            logger.info(f"tofu apply completed with exit code {result.returncode}")
+            logger.info(f"tofu apply completed with exit code {returncode}")
 
             # Capture outputs if apply succeeded
             outputs = {}
-            if result.returncode == 0:
+            if returncode == 0:
                 outputs = self._capture_outputs(work_dir, tofu_env)
                 if module is not None:
                     normalized = normalize_infrastructure_access_outputs(
@@ -1717,7 +1819,7 @@ def run_apply(
                     )
                     outputs = normalized.outputs
 
-            return result.returncode, output, outputs
+            return returncode, output, outputs
 
         except subprocess.TimeoutExpired as e:
             # S14-021: Handle timeout for apply
@@ -1841,7 +1943,13 @@ def normalize_outputs(self, outputs: dict, module: ProjectModule) -> dict:
         )
         return normalized.outputs
 
-    def run_destroy(self, work_dir: str, env: dict, timeout: int | None = None) -> tuple[int, str]:
+    def run_destroy(
+        self,
+        work_dir: str,
+        env: dict,
+        timeout: int | None = None,
+        on_output: Callable[[str], None] | None = None,
+    ) -> tuple[int, str]:
         """
         Run tofu destroy.
 
@@ -1849,6 +1957,7 @@ def run_destroy(self, work_dir: str, env: dict, timeout: int | None = None) -> t
             work_dir: Workspace directory
             env: Environment variables
             timeout: Timeout in seconds (default: 30 minutes)
+            on_output: Optional line callback for incremental log streaming (#195).
 
         Returns:
             Tuple of (exit_code, output)
@@ -1860,19 +1969,14 @@ def run_destroy(self, work_dir: str, env: dict, timeout: int | None = None) -> t
         logger.info(f"Running tofu destroy in {work_dir} (timeout: {timeout}s)")
 
         try:
-            result = subprocess.run(
+            returncode, output = self._run_tofu(
                 ["tofu", "destroy", "-no-color", "-input=false", "-auto-approve"],
-                cwd=work_dir,
-                env=tofu_env,
-                capture_output=True,
-                text=True,
-                timeout=timeout
+                work_dir, tofu_env, timeout, on_output,
             )
 
-            output = result.stdout + result.stderr
-            logger.info(f"tofu destroy completed with exit code {result.returncode}")
+            logger.info(f"tofu destroy completed with exit code {returncode}")
 
-            return result.returncode, output
+            return returncode, output
 
         except subprocess.TimeoutExpired as e:
             timeout_msg = (
@@ -1895,7 +1999,8 @@ def run_destroy_with_retry(
         module: ProjectModule | None = None,
         max_retries: int | None = None,
         initial_delay: float | None = None,
-        timeout: int | None = None
+        timeout: int | None = None,
+        on_output: Callable[[str], None] | None = None,
     ) -> tuple[int, str]:
         """
         Run tofu destroy with retry logic for dependency violations.
@@ -1941,7 +2046,9 @@ def run_destroy_with_retry(
                 time.sleep(delay)
 
             # Run destroy
-            exit_code, output = self.run_destroy(work_dir, env, timeout=timeout)
+            exit_code, output = self.run_destroy(
+                work_dir, env, timeout=timeout, on_output=on_output,
+            )
             all_output += output
 
             # Success - return immediately
diff --git a/backend/tasks/_tofu_helpers.py b/backend/tasks/_tofu_helpers.py
index aee67b3..ac0991b 100644
--- a/backend/tasks/_tofu_helpers.py
+++ b/backend/tasks/_tofu_helpers.py
@@ -11,6 +11,8 @@
 
 import logging
 import re
+import time
+from collections.abc import Callable
 from datetime import UTC, datetime
 
 from celery import Task
@@ -23,6 +25,67 @@
 logger = logging.getLogger(__name__)
 
 
+class TofuLogStreamer:
+    """Incrementally flush ``task.logs`` to the DB as a tofu step streams output.
+
+    Fixes the first defect in #195: OpenTofu task logs were buffered until the
+    task completed, so ``logs_full_size`` sat at 0 for the whole run and only
+    jumped to its final value at the end — a running module was opaque. The
+    ``run_*`` runtime methods now stream stdout line-by-line through an
+    ``on_output`` callback; this helper turns that callback into throttled
+    writes of ``task.logs`` so the task-detail endpoint's ``logs_full_size``
+    grows *during* the run.
+
+    Usage::
+
+        streamer = TofuLogStreamer(task, db)
+        ...
+        all_logs += header
+        code, logs = engine.run_plan(work_dir, env, on_output=streamer.begin(all_logs))
+        all_logs += logs                 # unchanged: final source of truth
+        task.logs = all_logs             # written + committed by the task as before
+
+    ``begin(base)`` snapshots the log accumulated before the step (section
+    headers + earlier steps) and returns the sink. During the step the sink
+    persists ``base + <lines streamed so far>``. The task still writes the
+    complete ``all_logs`` (built from each step's returned output) at the end,
+    so the final content and size are exactly what they were before — only the
+    *timing* of visibility changes. The sink runs on the task's own thread (the
+    runtime reads the pipe synchronously), so touching ``task``/``db`` here is
+    safe. Both the write and commit are best-effort: a log flush must never
+    fail the operation.
+    """
+
+    def __init__(self, task: TaskModel, db, *, interval: float = 2.0):
+        self._task = task
+        self._db = db
+        self._interval = interval
+        self._base = ""
+        self._buf: list[str] = []
+        self._last = 0.0
+
+    def begin(self, base: str) -> Callable[[str], None]:
+        """Start streaming a step whose output extends ``base``; return the sink."""
+        self._base = base
+        self._buf = []
+        self._last = 0.0  # force the first line to flush immediately
+        return self._sink
+
+    def _sink(self, line: str) -> None:
+        self._buf.append(line + "\n")
+        now = time.monotonic()
+        if now - self._last >= self._interval:
+            self._last = now
+            self._flush()
+
+    def _flush(self) -> None:
+        try:
+            self._task.logs = self._base + "".join(self._buf)
+            self._db.commit()
+        except Exception:  # noqa: BLE001 — a log flush must never fail the step
+            self._db.rollback()
+
+
 # ============================================================================
 # DRY Helper Functions
 # ============================================================================
diff --git a/backend/tasks/opentofu_tasks.py b/backend/tasks/opentofu_tasks.py
index dc79711..2eeb188 100644
--- a/backend/tasks/opentofu_tasks.py
+++ b/backend/tasks/opentofu_tasks.py
@@ -38,6 +38,7 @@
 from tasks._task_lookup import fetch_task_or_raise
 from tasks._tofu_helpers import (
     CallbackTask,
+    TofuLogStreamer,
     _cleanup_stuck_finalizers,
     _create_notification,
     _is_namespace_finalizer_issue,
@@ -244,8 +245,12 @@ def run_opentofu_init(self, task_db_id: int, module_id: int, keep_workspace: boo
                 # Get credentials
                 env = get_cloud_credentials_env(project, db)
 
-                # Run init
-                exit_code, logs = engine.run_init(work_dir, env)
+                # Run init — stream output so logs_full_size grows during the
+                # run instead of only at completion (#195).
+                streamer = TofuLogStreamer(task, db)
+                exit_code, logs = engine.run_init(
+                    work_dir, env, on_output=streamer.begin(""),
+                )
 
                 # Update task
                 task.exit_code = exit_code
@@ -424,6 +429,8 @@ def run_opentofu_plan(self, task_db_id: int, module_id: int, keep_workspace: boo
                 db.commit()
 
                 all_logs = ""
+                # Stream tofu output into task.logs during the run (#195).
+                streamer = TofuLogStreamer(task, db)
 
                 # Check if workspace is initialized - run init if needed
                 needs_reinit, reinit_reason = workspace.needs_reinit(module)
@@ -438,7 +445,9 @@ def run_opentofu_plan(self, task_db_id: int, module_id: int, keep_workspace: boo
                     # Need to run init first
                     task.command = "tofu init && tofu plan"
                     all_logs += f"[{_ts()}] --- INIT ---\n"
-                    init_code, init_logs = engine.run_init(work_dir, env)
+                    init_code, init_logs = engine.run_init(
+                        work_dir, env, on_output=streamer.begin(all_logs),
+                    )
                     all_logs += init_logs
 
                     if init_code != 0:
@@ -458,7 +467,9 @@ def run_opentofu_plan(self, task_db_id: int, module_id: int, keep_workspace: boo
 
                 # Run plan (saves plan.out to workspace)
                 all_logs += f"[{_ts()}] --- PLAN ---\n"
-                exit_code, plan_logs = engine.run_plan(work_dir, env)
+                exit_code, plan_logs = engine.run_plan(
+                    work_dir, env, on_output=streamer.begin(all_logs),
+                )
                 all_logs += plan_logs
 
                 task.exit_code = exit_code
@@ -603,6 +614,8 @@ def run_opentofu_apply(self, task_db_id: int, module_id: int, keep_workspace: bo
 
                 all_logs = ""
                 used_saved_plan = False
+                # Stream tofu output into task.logs during the run (#195).
+                streamer = TofuLogStreamer(task, db)
 
                 def _ensure_workspace_initialized() -> bool:
                     """Ensure providers/modules are installed before reconcile/plan/apply."""
@@ -616,7 +629,9 @@ def _ensure_workspace_initialized() -> bool:
                         all_logs += f"[{_ts()}] === INIT REQUIRED: {reinit_reason} ===\n"
 
                     all_logs += f"[{_ts()}] --- INIT ---\n"
-                    init_code, init_logs = engine.run_init(work_dir, env)
+                    init_code, init_logs = engine.run_init(
+                        work_dir, env, on_output=streamer.begin(all_logs),
+                    )
                     all_logs += init_logs
                     if init_code != 0:
                         task.exit_code = init_code
@@ -705,7 +720,9 @@ def _ensure_workspace_initialized() -> bool:
 
                     # Run plan
                     all_logs += f"[{_ts()}] --- PLAN ---\n"
-                    plan_code, plan_logs = engine.run_plan(work_dir, env)
+                    plan_code, plan_logs = engine.run_plan(
+                        work_dir, env, on_output=streamer.begin(all_logs),
+                    )
                     all_logs += plan_logs
                     if plan_code != 0:
                         task.exit_code = plan_code
@@ -725,7 +742,9 @@ def _ensure_workspace_initialized() -> bool:
 
                 # Apply (uses plan.out which exists either from saved plan or just-created plan)
                 all_logs += f"[{_ts()}] --- APPLY ---\n"
-                apply_code, apply_logs, outputs = engine.run_apply(work_dir, env, module=module)
+                apply_code, apply_logs, outputs = engine.run_apply(
+                    work_dir, env, module=module, on_output=streamer.begin(all_logs),
+                )
 
                 # Bounded stale-plan recovery: clear stale plan, re-plan once, then retry apply.
                 # This prevents repeated failures when remote state changed after plan creation.
@@ -746,7 +765,9 @@ def _ensure_workspace_initialized() -> bool:
                     else:
                         if reinit_reason:
                             all_logs += f"[{_ts()}] === INIT REQUIRED: {reinit_reason} ===\n"
-                        init_code, init_logs = engine.run_init(work_dir, env)
+                        init_code, init_logs = engine.run_init(
+                            work_dir, env, on_output=streamer.begin(all_logs),
+                        )
                         all_logs += init_logs
                         if init_code != 0:
                             task.exit_code = init_code
@@ -763,7 +784,9 @@ def _ensure_workspace_initialized() -> bool:
                         all_logs += "\n"
 
                     all_logs += f"[{_ts()}] --- PLAN (RETRY AFTER STALE PLAN) ---\n"
-                    plan_code, plan_logs = engine.run_plan(work_dir, env)
+                    plan_code, plan_logs = engine.run_plan(
+                        work_dir, env, on_output=streamer.begin(all_logs),
+                    )
                     all_logs += plan_logs
                     if plan_code != 0:
                         task.exit_code = plan_code
@@ -781,7 +804,9 @@ def _ensure_workspace_initialized() -> bool:
 
                     all_logs += "\n"
                     all_logs += f"[{_ts()}] --- APPLY (RETRY) ---\n"
-                    apply_code, apply_logs, outputs = engine.run_apply(work_dir, env, module=module)
+                    apply_code, apply_logs, outputs = engine.run_apply(
+                        work_dir, env, module=module, on_output=streamer.begin(all_logs),
+                    )
 
                 all_logs += apply_logs
 
@@ -1041,6 +1066,8 @@ def run_opentofu_destroy(self, task_db_id: int, module_id: int, keep_workspace:
                 db.commit()
 
                 all_logs = ""
+                # Stream tofu output into task.logs during the run (#195).
+                streamer = TofuLogStreamer(task, db)
 
                 # Check if workspace is initialized
                 needs_reinit, reinit_reason = workspace.needs_reinit(module)
@@ -1053,7 +1080,9 @@ def run_opentofu_destroy(self, task_db_id: int, module_id: int, keep_workspace:
                     task.command = "tofu init && tofu destroy"
                     # Init
                     all_logs += f"[{_ts()}] --- INIT ---\n"
-                    init_code, init_logs = engine.run_init(work_dir, env)
+                    init_code, init_logs = engine.run_init(
+                        work_dir, env, on_output=streamer.begin(all_logs),
+                    )
                     all_logs += init_logs
                     if init_code != 0:
                         task.exit_code = init_code
@@ -1073,7 +1102,8 @@ def run_opentofu_destroy(self, task_db_id: int, module_id: int, keep_workspace:
                 all_logs += f"[{_ts()}] --- DESTROY ---\n"
                 timeout = engine.get_destroy_timeout(module)
                 destroy_code, destroy_logs = engine.run_destroy_with_retry(
-                    work_dir, env, module=module, timeout=timeout
+                    work_dir, env, module=module, timeout=timeout,
+                    on_output=streamer.begin(all_logs),
                 )
                 all_logs += destroy_logs
 
@@ -1090,7 +1120,8 @@ def run_opentofu_destroy(self, task_db_id: int, module_id: int, keep_workspace:
                         logger.info("Retrying destroy after finalizer cleanup")
                         all_logs += f"\n[{_ts()}] --- DESTROY RETRY ---\n"
                         retry_code, retry_logs = engine.run_destroy_with_retry(
-                            work_dir, env, module=module, timeout=300  # Shorter timeout for retry
+                            work_dir, env, module=module, timeout=300,  # Shorter timeout for retry
+                            on_output=streamer.begin(all_logs),
                         )
                         all_logs += f"{retry_logs}\n"
 
diff --git a/backend/tests/component/test_opentofu_runtime_streaming.py b/backend/tests/component/test_opentofu_runtime_streaming.py
new file mode 100644
index 0000000..3bd232a
--- /dev/null
+++ b/backend/tests/component/test_opentofu_runtime_streaming.py
@@ -0,0 +1,195 @@
+"""#195: OpenTofu runtime streams subprocess output incrementally.
+
+Before this fix ``run_init``/``run_plan``/``run_apply``/``run_destroy`` used a
+blocking ``subprocess.run(capture_output=True)`` that returned the entire log
+only when the process exited, so a caller had nothing to persist until the very
+end. These tests prove the streaming path delivers each line *while the process
+is still running*, and that the ``run_*`` methods route through it only when an
+``on_output`` callback is supplied (the classic blocking capture is preserved
+otherwise, which the existing test_opentofu_runtime.py suite locks).
+"""
+
+import os
+import subprocess
+import sys
+
+import pytest
+
+import services.execution.opentofu_runtime as otr
+from services.execution.opentofu_runtime import OpenTofuRuntime, _stream_subprocess
+
+
+@pytest.mark.component
+class TestStreamSubprocess:
+    def test_delivers_each_line_before_process_finishes(self, tmp_path):
+        """A handshake proves lines arrive live, not buffered until exit.
+
+        The child prints ``line-1`` then blocks on a sentinel that the test's
+        ``on_output`` writes only when it *receives* ``line-1``; only then does
+        the child print ``line-2``. The blocking loop runs FAR longer than the
+        watchdog (``timeout`` below), so a buffered implementation — the #195 bug,
+        where ``on_output`` fires only after the process exits — cannot
+        self-release: the sentinel never appears, the child deadlocks, and the
+        watchdog kills it → ``_stream_subprocess`` raises ``TimeoutExpired`` and
+        this test ERRORS. Genuine streaming releases the child within
+        milliseconds, so the call returns well under the watchdog. Both the raise
+        AND the timing bound below make the distinction non-vacuous — a fully
+        buffered impl fails, proven by mutation.
+        """
+        import time
+
+        go = tmp_path / "go"
+        # ~300s of blocking: >> the 8s watchdog, so a buffered impl MUST deadlock
+        # rather than self-release before the watchdog fires.
+        script = (
+            "import sys, time, pathlib\n"
+            "print('line-1', flush=True)\n"
+            "sentinel = pathlib.Path(sys.argv[1])\n"
+            "for _ in range(6000):\n"
+            "    if sentinel.exists():\n"
+            "        break\n"
+            "    time.sleep(0.05)\n"
+            "print('line-2', flush=True)\n"
+        )
+
+        received: list[str] = []
+
+        def on_output(line: str) -> None:
+            received.append(line)
+            if line == "line-1":
+                go.write_text("go")  # unblock the child only after we SEE line-1
+
+        started = time.monotonic()
+        code, output = _stream_subprocess(
+            [sys.executable, "-c", script, str(go)],
+            cwd=str(tmp_path),
+            env=dict(os.environ),
+            timeout=8,
+            on_output=on_output,
+        )
+        elapsed = time.monotonic() - started
+
+        assert code == 0
+        # line-2 was printed *only* because on_output saw line-1 and released it.
+        assert received == ["line-1", "line-2"]
+        assert "line-1" in output and "line-2" in output
+        # Live streaming releases the child in ms; a buffered impl would deadlock
+        # and blow the 8s watchdog. The timing bound makes that explicit.
+        assert elapsed < 5, f"took {elapsed:.1f}s — output looks buffered, not streamed"
+
+    def test_returns_combined_output_and_exit_code(self, tmp_path):
+        script = "import sys; print('out'); print('err', file=sys.stderr); sys.exit(3)"
+        seen: list[str] = []
+        code, output = _stream_subprocess(
+            [sys.executable, "-c", script],
+            cwd=str(tmp_path),
+            env=dict(os.environ),
+            timeout=20,
+            on_output=seen.append,
+        )
+        assert code == 3
+        # stderr is merged into stdout so both surface in the log and the sink.
+        assert "out" in output and "err" in output
+        assert "out" in seen and "err" in seen
+
+    def test_timeout_kills_and_raises_with_partial_output(self, tmp_path):
+        script = (
+            "import sys, time\n"
+            "print('before-hang', flush=True)\n"
+            "time.sleep(30)\n"
+        )
+        seen: list[str] = []
+        with pytest.raises(subprocess.TimeoutExpired) as exc:
+            _stream_subprocess(
+                [sys.executable, "-c", script],
+                cwd=str(tmp_path),
+                env=dict(os.environ),
+                timeout=1,
+                on_output=seen.append,
+            )
+        # The partial output produced before the hang is carried on the
+        # exception, matching subprocess.run(...).stdout semantics the run_*
+        # timeout branches rely on.
+        assert "before-hang" in (exc.value.output or "")
+        assert seen == ["before-hang"]
+
+
+@pytest.mark.component
+class TestRunMethodsRouteThroughStreaming:
+    @staticmethod
+    def _runtime(monkeypatch):
+        # OpenTofuRuntime only needs a DB session for workspace ops, not for the
+        # subprocess-running methods under test here. Stub the system-defaults
+        # gate so construction doesn't need a real DB.
+        monkeypatch.setattr(
+            otr, "check_required_configured",
+            lambda _db: {"all_configured": True, "missing": []},
+        )
+        return OpenTofuRuntime(db=None)
+
+    def _patch_stream(self, monkeypatch):
+        calls: dict = {}
+
+        def fake_stream(cmd, *, cwd, env, timeout, on_output):
+            calls["cmd"] = cmd
+            calls["on_output"] = on_output
+            on_output("streamed-1")
+            on_output("streamed-2")
+            return 0, "streamed-1\nstreamed-2\n"
+
+        monkeypatch.setattr(otr, "_stream_subprocess", fake_stream)
+        return calls
+
+    def test_run_plan_streams_when_on_output_given(self, monkeypatch):
+        calls = self._patch_stream(monkeypatch)
+        runtime = self._runtime(monkeypatch)
+        seen: list[str] = []
+
+        code, output = runtime.run_plan("/tmp/w", {}, on_output=seen.append)
+
+        assert code == 0
+        assert seen == ["streamed-1", "streamed-2"]
+        assert "streamed-1" in output and "streamed-2" in output
+        assert "plan" in calls["cmd"]
+
+    def test_run_init_streams_when_on_output_given(self, monkeypatch):
+        calls = self._patch_stream(monkeypatch)
+        runtime = self._runtime(monkeypatch)
+        seen: list[str] = []
+
+        code, output = runtime.run_init("/tmp/w", {}, on_output=seen.append)
+
+        assert code == 0
+        assert seen == ["streamed-1", "streamed-2"]
+        assert "init" in calls["cmd"]
+
+    def test_run_apply_streams_when_on_output_given(self, monkeypatch):
+        self._patch_stream(monkeypatch)
+        runtime = self._runtime(monkeypatch)
+        # apply captures outputs on success; stub that out.
+        monkeypatch.setattr(runtime, "_capture_outputs", lambda *a, **k: {})
+        seen: list[str] = []
+
+        code, output, _outputs = runtime.run_apply("/tmp/w", {}, on_output=seen.append)
+
+        assert code == 0
+        assert seen == ["streamed-1", "streamed-2"]
+
+    def test_blocking_path_used_when_no_on_output(self, monkeypatch):
+        """Without on_output the classic subprocess.run capture is used, NOT the
+        streaming helper — preserving legacy behaviour and the existing suite."""
+        def _boom(*a, **k):
+            raise AssertionError("_stream_subprocess must not run without on_output")
+
+        monkeypatch.setattr(otr, "_stream_subprocess", _boom)
+
+        completed = subprocess.CompletedProcess(
+            args=["tofu", "plan"], returncode=0, stdout="blocking-out", stderr="",
+        )
+        monkeypatch.setattr(otr.subprocess, "run", lambda *a, **k: completed)
+
+        runtime = self._runtime(monkeypatch)
+        code, output = runtime.run_plan("/tmp/w", {})
+
+        assert code == 0
+        assert output == "blocking-out"
diff --git a/backend/tests/component/test_opentofu_tasks.py b/backend/tests/component/test_opentofu_tasks.py
index 81e89fd..c29fe5b 100644
--- a/backend/tests/component/test_opentofu_tasks.py
+++ b/backend/tests/component/test_opentofu_tasks.py
@@ -543,7 +543,12 @@ def test_apply_reconcile_imports_clear_saved_plan_before_plan_reuse(
         assert result["success"] is True
         assert mock_workspace.clear_plan.call_count == 2
         assert all(call.args == (module,) for call in mock_workspace.clear_plan.call_args_list)
-        mock_engine.run_plan.assert_called_once_with("/tmp/workspace", {})
+        # run_plan is now called with an additive on_output= streaming callback
+        # (#195); assert on the positional args and ignore the sink kwarg.
+        mock_engine.run_plan.assert_called_once()
+        plan_args, plan_kwargs = mock_engine.run_plan.call_args
+        assert plan_args == ("/tmp/workspace", {})
+        assert set(plan_kwargs) <= {"on_output"}
 
         db.refresh(task)
         assert "=== RECONCILIATION INVALIDATED SAVED PLAN ===" in (task.logs or "")
@@ -719,3 +724,148 @@ def test_roks_register_enqueues_scan(self, db):
             enqueue_cluster_scan(cluster.id)
 
         mock_scan_task.delay.assert_called_once_with(55)
+
+
+# ── #195: Incremental log streaming during a run ─────────────────────────────
+
+class TestOpenTofuTaskLogStreaming:
+    """logs_full_size must grow while the task is in_progress, not jump from 0
+    to its final value only at completion (issue #195)."""
+
+    @patch(f"{_MOD}.update_project_counts")
+    @patch(f"{_MOD}.create_deployment_record")
+    @patch(f"{_MOD}._notify_task_started")
+    @patch(f"{_MOD}.module_lock")
+    @patch(f"{_MOD}.get_cloud_credentials_env", return_value={})
+    @patch(f"{_MOD}.check_dependencies", return_value=(True, []))
+    @patch(f"{_MOD}.OpenTofuRuntime")
+    @patch(f"{_MOD}.get_db_context")
+    @patch(f"{_MOD}.datetime")
+    def test_plan_persists_growing_logs_before_completion(
+        self, mock_dt, mock_db_ctx, mock_runtime_cls, _mock_deps, _mock_creds,
+        mock_lock, _mock_notify, _mock_deploy, _mock_counts, db,
+    ):
+        from sqlalchemy import func
+
+        from tasks._tofu_helpers import TofuLogStreamer
+
+        naive_now = datetime.utcnow()
+        mock_dt.now.return_value = naive_now
+        mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
+
+        project = _make_project(db)
+        lib = _make_library(db)
+        module = _make_module(db, project, lib, status="initialized")
+        task = _make_task(db, project, module, "plan")
+        db.commit()
+
+        mock_db_ctx.return_value.__enter__ = MagicMock(return_value=db)
+        mock_db_ctx.return_value.__exit__ = MagicMock(return_value=False)
+        mock_lock.return_value.__enter__ = MagicMock(
+            return_value=ModuleLock(module_id=module.id, task_id=task.id, fence_token=0),
+        )
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        # As each line streams, record what a concurrent GET /api/tasks/{id}
+        # would observe: the persisted logs_full_size (SQL length, exactly the
+        # detail endpoint's computation) and the task status. This is measured
+        # from the DB, not the in-memory buffer, so it proves the flush.
+        observed_sizes: list[int] = []
+        observed_status: list[str] = []
+
+        def streaming_run_plan(work_dir, env, on_output=None, timeout=None):
+            assert on_output is not None, "plan task must pass an on_output sink (#195)"
+            for i in range(1, 6):
+                on_output(f"tofu progress line {i}")
+                size = db.query(func.length(TaskModel.logs)).filter(
+                    TaskModel.id == task.id
+                ).scalar()
+                status = db.query(TaskModel.status).filter(
+                    TaskModel.id == task.id
+                ).scalar()
+                observed_sizes.append(size or 0)
+                observed_status.append(status)
+            return (0, "".join(f"tofu progress line {i}\n" for i in range(1, 6)))
+
+        mock_engine = MagicMock()
+        mock_engine.prepare_persistent_workspace.return_value = "/tmp/workspace"
+        mock_engine.run_plan.side_effect = streaming_run_plan
+        mock_runtime_cls.return_value = mock_engine
+
+        mock_workspace = MagicMock()
+        mock_workspace.is_initialized.return_value = True
+        mock_workspace.needs_reinit.return_value = (False, None)
+
+        # Force interval=0 so every streamed line flushes — otherwise the 2s
+        # throttle would coalesce this fast test's lines into a single flush.
+        def _fast_streamer(t, d, **_kw):
+            return TofuLogStreamer(t, d, interval=0)
+
+        with patch("services.workspace_manager.WorkspaceManager", return_value=mock_workspace), \
+             patch(f"{_MOD}.TofuLogStreamer", _fast_streamer):
+            from tasks.opentofu_tasks import run_opentofu_plan
+            result = run_opentofu_plan(task.id, module.id)
+
+        assert result["success"] is True
+
+        # (a) The bug reproduced: without streaming every one of these would be
+        #     0. Instead the size is non-zero from the first line and grows.
+        assert len(observed_sizes) == 5
+        assert all(s > 0 for s in observed_sizes), observed_sizes
+        assert observed_sizes == sorted(observed_sizes)
+        assert len(set(observed_sizes)) == 5, f"not strictly growing: {observed_sizes}"
+        # (b) …and this all happened while the task was still running.
+        assert observed_status == ["in_progress"] * 5
+
+        # Final state intact: complete log persisted, size == detail's len().
+        db.refresh(task)
+        assert task.status == "completed"
+        assert "tofu progress line 5" in task.logs
+        assert len(task.logs) >= observed_sizes[-1]
+
+    @patch(f"{_MOD}.update_project_counts")
+    @patch(f"{_MOD}.create_deployment_record")
+    @patch(f"{_MOD}._notify_task_started")
+    @patch(f"{_MOD}.module_lock")
+    @patch(f"{_MOD}.get_cloud_credentials_env", return_value={})
+    @patch(f"{_MOD}.check_dependencies", return_value=(True, []))
+    @patch(f"{_MOD}.OpenTofuRuntime")
+    @patch(f"{_MOD}.get_db_context")
+    @patch(f"{_MOD}.datetime")
+    def test_plan_passes_on_output_sink_to_runtime(
+        self, mock_dt, mock_db_ctx, mock_runtime_cls, _mock_deps, _mock_creds,
+        mock_lock, _mock_notify, _mock_deploy, _mock_counts, db,
+    ):
+        """Guard: the plan task must hand run_plan a callable on_output sink."""
+        naive_now = datetime.utcnow()
+        mock_dt.now.return_value = naive_now
+        mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
+
+        project = _make_project(db)
+        lib = _make_library(db)
+        module = _make_module(db, project, lib, status="initialized")
+        task = _make_task(db, project, module, "plan")
+        db.commit()
+
+        mock_db_ctx.return_value.__enter__ = MagicMock(return_value=db)
+        mock_db_ctx.return_value.__exit__ = MagicMock(return_value=False)
+        mock_lock.return_value.__enter__ = MagicMock(
+            return_value=ModuleLock(module_id=module.id, task_id=task.id, fence_token=0),
+        )
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        mock_engine = MagicMock()
+        mock_engine.prepare_persistent_workspace.return_value = "/tmp/workspace"
+        mock_engine.run_plan.return_value = (0, "Plan: 1 to add")
+        mock_runtime_cls.return_value = mock_engine
+
+        mock_workspace = MagicMock()
+        mock_workspace.is_initialized.return_value = True
+        mock_workspace.needs_reinit.return_value = (False, None)
+
+        with patch("services.workspace_manager.WorkspaceManager", return_value=mock_workspace):
+            from tasks.opentofu_tasks import run_opentofu_plan
+            run_opentofu_plan(task.id, module.id)
+
+        _args, kwargs = mock_engine.run_plan.call_args
+        assert callable(kwargs.get("on_output"))
diff --git a/backend/tests/integration/test_routes_tasks.py b/backend/tests/integration/test_routes_tasks.py
index 69b85d9..ca1a4c1 100644
--- a/backend/tests/integration/test_routes_tasks.py
+++ b/backend/tests/integration/test_routes_tasks.py
@@ -49,6 +49,58 @@ def test_list_tasks_with_status_filter(self, client, admin_headers, sample_user,
             assert task["status"] == "completed"
 
 
+class TestTaskListLogFields:
+    """#195 (second defect): the list endpoint must expose logs_full_size so a
+    client enumerating a module's tasks can tell empty from populated. It used
+    to be omitted entirely (surfaced as null), while the detail endpoint had it."""
+
+    def test_list_includes_logs_full_size_matching_detail(
+        self, client, admin_headers, sample_user, sample_project, db
+    ):
+        from tests.factories import TaskFactory
+
+        log_body = "\n".join(f"[04:42:{i:02d}] line {i}" for i in range(20))
+        task = TaskFactory(
+            db, project=sample_project, task_type="plan",
+            status="completed", logs=log_body,
+        )
+        db.commit()
+
+        listed = client.get(
+            f"/api/tasks?project_id={sample_project.id}", headers=admin_headers
+        ).json()
+        row = next(t for t in listed["tasks"] if t["id"] == task.id)
+
+        # The field is present (not omitted) and equals the true log size.
+        assert "logs_full_size" in row
+        assert row["logs_full_size"] == len(log_body)
+        assert "logs_truncated" in row
+
+        # …and it matches what the detail endpoint reports for the same task.
+        detail = client.get(f"/api/tasks/{task.id}", headers=admin_headers).json()
+        assert detail["logs_full_size"] == row["logs_full_size"]
+
+    def test_list_reports_zero_for_task_without_logs(
+        self, client, admin_headers, sample_user, sample_project, db
+    ):
+        from tests.factories import TaskFactory
+
+        task = TaskFactory(
+            db, project=sample_project, task_type="plan", status="queued", logs=None
+        )
+        db.commit()
+
+        listed = client.get(
+            f"/api/tasks?project_id={sample_project.id}", headers=admin_headers
+        ).json()
+        row = next(t for t in listed["tasks"] if t["id"] == task.id)
+
+        # NULL logs → 0 (matching the detail endpoint), not null/omitted.
+        assert row["logs_full_size"] == 0
+        detail = client.get(f"/api/tasks/{task.id}", headers=admin_headers).json()
+        assert detail["logs_full_size"] == 0
+
+
 class TestTaskDetail:
     """GET /api/tasks/{id}."""
 
diff --git a/backend/tests/unit/test_tofu_log_streamer.py b/backend/tests/unit/test_tofu_log_streamer.py
new file mode 100644
index 0000000..fc1c498
--- /dev/null
+++ b/backend/tests/unit/test_tofu_log_streamer.py
@@ -0,0 +1,85 @@
+"""Unit tests for TofuLogStreamer — the incremental task.logs flusher (#195).
+
+The streamer turns the runtime's per-line ``on_output`` callback into throttled
+writes of ``task.logs`` so ``logs_full_size`` grows during a run. These tests
+lock: base+buffer composition, growth across lines, interval throttling, the
+force-flush of the first line, and that a failed commit never propagates.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from tasks._tofu_helpers import TofuLogStreamer
+
+
+@pytest.mark.unit
+class TestTofuLogStreamer:
+    def test_first_line_flushes_and_prepends_base(self):
+        task = MagicMock()
+        db = MagicMock()
+        streamer = TofuLogStreamer(task, db, interval=0)
+
+        sink = streamer.begin("HEADER\n")
+        sink("first line")
+
+        assert task.logs == "HEADER\nfirst line\n"
+        db.commit.assert_called()
+
+    def test_logs_grow_across_lines(self):
+        task = MagicMock()
+        db = MagicMock()
+        streamer = TofuLogStreamer(task, db, interval=0)  # every line flushes
+
+        sink = streamer.begin("BASE\n")
+        sizes = []
+        for i in range(1, 6):
+            sink(f"line {i}")
+            sizes.append(len(task.logs))
+
+        # Strictly increasing — the whole point of #195: not 0 until the end.
+        assert sizes == sorted(sizes)
+        assert len(set(sizes)) == len(sizes)
+        assert db.commit.call_count == 5
+        assert task.logs.startswith("BASE\n")
+        assert "line 5" in task.logs
+
+    def test_interval_throttles_intermediate_lines(self, monkeypatch):
+        task = MagicMock()
+        db = MagicMock()
+        # begin() sets last=0.0 so the first line always flushes; a large
+        # interval then suppresses the second line arriving within the window.
+        monkeypatch.setattr("time.monotonic", MagicMock(side_effect=[1000.0, 1000.1]))
+        streamer = TofuLogStreamer(task, db, interval=100)
+
+        sink = streamer.begin("")
+        sink("first")   # 1000.0 - 0.0 >= 100 -> flush
+        sink("second")  # 1000.1 - 1000.0 < 100 -> throttled
+
+        assert db.commit.call_count == 1
+        # Persisted content still reflects only the flushed line; the throttled
+        # line is folded into task.logs by the task's final all_logs write.
+        assert task.logs == "first\n"
+
+    def test_begin_resets_base_and_buffer(self):
+        task = MagicMock()
+        db = MagicMock()
+        streamer = TofuLogStreamer(task, db, interval=0)
+
+        streamer.begin("STEP-A\n")("a-line")
+        assert task.logs == "STEP-A\na-line\n"
+
+        # A new step starts from a fresh base; the previous step's buffer is gone.
+        streamer.begin("STEP-A\na-line\nSTEP-B\n")("b-line")
+        assert task.logs == "STEP-A\na-line\nSTEP-B\nb-line\n"
+
+    def test_commit_failure_is_swallowed_and_rolls_back(self):
+        task = MagicMock()
+        db = MagicMock()
+        db.commit.side_effect = RuntimeError("db gone")
+        streamer = TofuLogStreamer(task, db, interval=0)
+
+        sink = streamer.begin("H\n")
+        sink("a line")  # must not raise
+
+        db.rollback.assert_called_once()