diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0f8c09b..f1d6beb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,17 @@ All notable changes to DazScript Server are documented here.
 
 ## [Unreleased]
 
+### Structured async job observation
+
+Async script submissions may name a per-job `reportFile`. The server truncates
+it at submission, incrementally ingests its JSONL events while Daz's main thread
+is occupied, and exposes one `observation` record through status and result:
+structured progress, the most recent 100 log entries with truncation metadata,
+and a deduplicated output manifest. Request lists carry a bounded observation
+summary. Both `DazClient` and `AsyncDazClient` accept `report_file=` for inline
+and file-backed async jobs. Live report ingestion is available on Studio 6;
+Studio 4 parses the final report on the main thread after execution returns.
+
 ## [2.9.2] - 2026-08-21
 
 ### dazpy script-call batching
diff --git a/CLAUDE.md b/CLAUDE.md
index be7bc57..e039245 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -65,6 +65,9 @@ Synchronous execution crosses to `handleExecuteRequest()` via
 and validate with local reentrant Qt Core values, then submit to mutex-protected
 services without blocking on the main thread. Studio 4 must keep submission on
 the main thread because `JsonStd::parseObject()` uses `QScriptEngine` there.
+The same split applies to structured job reports: Studio 6 may ingest JSONL
+from polling workers for live observation, while Studio 4 parses the final
+report from `markCompleted()` on the main thread after script execution returns.
 Neither path may touch GUI objects, `DzScript`, the DAZ API, or shared mutable
 state from a worker. Settings used by an async handler are snapshotted before
 the server starts.
diff --git a/README.md b/README.md
index fd61800..0610b58 100644
--- a/README.md
+++ b/README.md
@@ -1772,7 +1772,33 @@ result = requests.get(f"{BASE}/requests/{req_id}/result?wait=true&timeout=300",
 **Purpose:** Submit an inline script or host-side script file for asynchronous execution
 **Authentication:** Required (if enabled)
 
-**Request Body:** Same as `POST /execute`
+**Request Body:** Same as `POST /execute`, plus optional `reportFile` for
+structured job observation.
+
+`reportFile` is an absolute host-side path to a JSONL file owned by this job.
+The server truncates it when the request is accepted. The script may then write
+one JSON object per line. The path is injected into the script's argument object
+as reserved field `getArguments()[0].__dssReportFile`, so callers do not have to
+duplicate it inside `args`:
+
+```jsonl
+{"type":"progress","value":0.25,"current":1,"total":4,"phase":"render","message":"Plate 1"}
+{"type":"log","level":"info","message":"Loaded scene"}
+{"type":"output","path":"C:/renders/run/plate-1.png","kind":"image","label":"plate"}
+```
+
+Progress events require either `value` (0–1) or `current` plus `total`. Log
+events require `message`. Output events require `path`; repeating a path updates
+that manifest entry. A `manifest` event may provide an `outputs` array. Use a
+unique report file for every job.
+`reportFile` may not be the submitted `scriptFile`, because the server owns and
+truncates the former before the job is queued.
+
+DAZ Studio 6 ingests report events from polling workers, so progress, logs, and
+outputs are visible while the main thread is occupied by the script. DAZ Studio
+4's JSON parser is main-thread-only; Studio 4 therefore ingests the final report
+after execution returns and does not expose report-driven progress while the job
+is still running.
 
 **Response (immediate):**
 ```json
@@ -1806,7 +1832,15 @@ result = requests.get(f"{BASE}/requests/{req_id}/result?wait=true&timeout=300",
 {
   "request_id": "a3f2b891",
   "status": "running",
-  "progress": 0.0,
+  "progress": 0.25,
+  "observation": {
+    "progress": {"value": 0.25, "current": 1, "total": 4, "phase": "render"},
+    "log_tail": [{"level": "info", "source": "script", "message": "Loaded scene"}],
+    "log_total": 1,
+    "log_truncated": false,
+    "output_manifest": {"count": 0, "outputs": []},
+    "report_file": "C:/renders/run/job.jsonl"
+  },
   "elapsed_ms": 1240,
   "queue_position": 0
 }
@@ -1844,6 +1878,11 @@ result = requests.get(f"{BASE}/requests/{req_id}/result?wait=true&timeout=300",
 }
 ```
 
+Completed, failed, and cancelled results also include `observation`. The log
+tail is capped at the newest 100 entries; `log_total` and `log_truncated` make
+the loss explicit. Ordinary captured `print()` output is folded into the same
+tail at completion while remaining available in the legacy `output` field.
+
 Returns HTTP 404 if the request ID is unknown or has been purged (TTL: 1 hour).
 
 ---
diff --git a/dazpy/_client.py b/dazpy/_client.py
index a89c28f..e0e6b94 100644
--- a/dazpy/_client.py
+++ b/dazpy/_client.py
@@ -262,13 +262,16 @@ def _do():
         return self._with_busy_retry(_do, retry_on_busy, max_wait)
 
     def execute_async_submit(
-        self, script: str, args: object = None, *, retry_on_busy: bool = False, max_wait: float = 30.0
+        self, script: str, args: object = None, *, report_file: str | None = None,
+        retry_on_busy: bool = False, max_wait: float = 30.0
     ) -> str:
         """Submit a script for asynchronous execution and return immediately.
 
         Args:
             script: DazScript source code.
             args: Optional argument for the script.
+            report_file: Optional host-side JSONL file used by the script to
+                report structured progress, logs, and output artefacts.
             retry_on_busy: If ``True``, transparently retry with backoff when
                 the server reports ``StudioBusyError``/``ConcurrencyLimitError``,
                 instead of raising immediately.
@@ -292,6 +295,8 @@ def execute_async_submit(
         payload: dict = {"script": script}
         if args is not None:
             payload["args"] = args
+        if report_file is not None:
+            payload["reportFile"] = report_file
 
         def _do():
             resp = self._post("/execute/async", payload)
@@ -301,7 +306,8 @@ def _do():
         return self._with_busy_retry(_do, retry_on_busy, max_wait)
 
     def execute_file_async_submit(
-        self, script_file: str, args: object = None, *, retry_on_busy: bool = False, max_wait: float = 30.0
+        self, script_file: str, args: object = None, *, report_file: str | None = None,
+        retry_on_busy: bool = False, max_wait: float = 30.0
     ) -> str:
         """Submit a host-side ``.dsa`` file for asynchronous execution.
 
@@ -313,6 +319,8 @@ def execute_file_async_submit(
         payload: dict = {"scriptFile": script_file}
         if args is not None:
             payload["args"] = args
+        if report_file is not None:
+            payload["reportFile"] = report_file
 
         def _do():
             resp = self._post("/execute/async", payload)
@@ -321,13 +329,17 @@ def _do():
 
         return self._with_busy_retry(_do, retry_on_busy, max_wait)
 
-    def execute_batch_async(self, operations: list[dict], args: object = None) -> str:
+    def execute_batch_async(
+        self, operations: list[dict], args: object = None, *, report_file: str | None = None
+    ) -> str:
         """Submit multiple operations as one async request (one queue slot, one script).
 
         Args:
             operations: List of ``{"body_lines": [...], "result_expression": "..."}``
                 dicts — same shape as :meth:`~dazpy.Batch.add_operation`'s arguments.
             args: Optional argument passed to the combined script.
+            report_file: Optional host-side JSONL file used for structured job
+                observation, as in :meth:`execute_async_submit`.
 
         Returns:
             The server-assigned ``request_id``. Poll it like any other async
@@ -338,7 +350,7 @@ def execute_batch_async(self, operations: list[dict], args: object = None) -> st
 
         pairs = [(op["body_lines"], op["result_expression"]) for op in operations]
         script = build_operations_script(pairs)
-        return self.execute_async_submit(script, args=args)
+        return self.execute_async_submit(script, args=args, report_file=report_file)
 
     def get_request_status(self, request_id: str) -> dict:
         """Return the current status of an async request.
@@ -347,7 +359,9 @@ def get_request_status(self, request_id: str) -> dict:
             request_id: The ID returned by :meth:`execute_async_submit`.
 
         Returns:
-            A dict with at least a ``"status"`` key.  Possible values:
+            A dict with at least ``"status"`` and ``"observation"`` keys.
+            Observation contains structured progress, the bounded log tail,
+            and the output manifest. Possible status values:
             ``"queued"``, ``"running"``, ``"completed"``, ``"failed"``,
             ``"cancelled"``, or ``"not_found"``.
         """
@@ -368,7 +382,7 @@ def get_request_result(self, request_id: str, wait: bool = False, wait_timeout:
 
         Returns:
             A dict containing ``success``, ``result``, ``output``, ``error``,
-            ``duration_ms``, and ``status`` keys.
+            ``duration_ms``, ``status``, and ``observation`` keys.
         """
         params = {}
         if wait:
diff --git a/dazpy/_client_aio.py b/dazpy/_client_aio.py
index 21756b7..dcd815b 100644
--- a/dazpy/_client_aio.py
+++ b/dazpy/_client_aio.py
@@ -127,12 +127,15 @@ async def _do():
         return await self._with_busy_retry(_do, retry_on_busy, max_wait)
 
     async def execute_async_submit(
-        self, script: str, args: object = None, *, retry_on_busy: bool = False, max_wait: float = 30.0
+        self, script: str, args: object = None, *, report_file: str | None = None,
+        retry_on_busy: bool = False, max_wait: float = 30.0
     ) -> str:
         """Submit a script for async execution. See :meth:`dazpy.DazClient.execute_async_submit`."""
         payload: dict = {"script": script}
         if args is not None:
             payload["args"] = args
+        if report_file is not None:
+            payload["reportFile"] = report_file
 
         async def _do():
             resp = await self._post("/execute/async", payload)
@@ -142,7 +145,8 @@ async def _do():
         return await self._with_busy_retry(_do, retry_on_busy, max_wait)
 
     async def execute_file_async_submit(
-        self, script_file: str, args: object = None, *, retry_on_busy: bool = False, max_wait: float = 30.0
+        self, script_file: str, args: object = None, *, report_file: str | None = None,
+        retry_on_busy: bool = False, max_wait: float = 30.0
     ) -> str:
         """Submit a host-side ``.dsa`` file asynchronously.
 
@@ -151,6 +155,8 @@ async def execute_file_async_submit(
         payload: dict = {"scriptFile": script_file}
         if args is not None:
             payload["args"] = args
+        if report_file is not None:
+            payload["reportFile"] = report_file
 
         async def _do():
             resp = await self._post("/execute/async", payload)
@@ -159,13 +165,17 @@ async def _do():
 
         return await self._with_busy_retry(_do, retry_on_busy, max_wait)
 
-    async def execute_batch_async(self, operations: list[dict], args: object = None) -> str:
+    async def execute_batch_async(
+        self, operations: list[dict], args: object = None, *, report_file: str | None = None
+    ) -> str:
         """Submit multiple operations as one async request. See :meth:`dazpy.DazClient.execute_batch_async`."""
         from ._batch import build_operations_script
 
         pairs = [(op["body_lines"], op["result_expression"]) for op in operations]
         script = build_operations_script(pairs)
-        return await self.execute_async_submit(script, args=args)
+        return await self.execute_async_submit(
+            script, args=args, report_file=report_file
+        )
 
     async def get_request_status(self, request_id: str) -> dict:
         """See :meth:`dazpy.DazClient.get_request_status`."""
diff --git a/include/AsyncRequestManager.h b/include/AsyncRequestManager.h
index d04c384..7c81073 100644
--- a/include/AsyncRequestManager.h
+++ b/include/AsyncRequestManager.h
@@ -7,6 +7,7 @@
 #include <QtCore/qpair.h>
 #include <QtCore/qvariant.h>
 #include <QtCore/qstringlist.h>
+#include <QtCore/qbytearray.h>
 #include <string>
 #include <utility>
 
@@ -34,6 +35,7 @@ class AsyncRequestManager {
 public:
     static const int DEFAULT_MAX_QUEUE_DEPTH      = 100;
     static const int DEFAULT_MAX_TRACKED_REQUESTS = 1000;
+    static const int DEFAULT_LOG_TAIL_LINES       = 100;
     static const int RESULT_POLL_INTERVAL_MS      = 500; // Long-poll sleep interval in getResultJson()
 
     // notifyTarget must be a DzScriptServerPane* (QObject subclass).
@@ -60,18 +62,22 @@ class AsyncRequestManager {
     // Enqueue a new async request. Returns SubmitResult::accepted=false when
     // the queue is at capacity or too many requests are tracked.
     SubmitResult        submit(const QString& scriptText, const QString& scriptFile,
-                               const QVariantMap& args, const QString& idPrefix);
+                               const QString& reportFile, const QVariantMap& args,
+                               const QString& idPrefix);
 
     // Enqueue a render job. Same as submit() but tags the request as
     // REQUEST_TYPE_RENDER so cancel dispatch calls killRender() correctly.
     SubmitResult        submitRender(const QString& scriptText, const QString& idPrefix);
 
-    // All four methods below are safe to call from HTTP threads (no Qt string ops).
-    std::pair<int, std::string> getStatusJson(const std::string& requestId) const;
+    // All five methods below are safe to call from HTTP threads. Studio 6 may
+    // ingest report JSON here because QJsonDocument is reentrant. Studio 4
+    // leaves report parsing to the main-thread completion path because its
+    // JsonStd parser constructs a thread-affine QScriptEngine.
+    std::pair<int, std::string> getStatusJson(const std::string& requestId);
     std::pair<int, std::string> getResultJson(const std::string& requestId, bool doWait, int timeoutSec);
     std::pair<int, std::string> cancelJson(const std::string& requestId, const std::string& clientIP);
     std::pair<int, std::string> cancelRenderJson(const std::string& requestId, const std::string& clientIP);
-    std::string                 listJson(const std::string& statusFilter) const;
+    std::string                 listJson(const std::string& statusFilter);
 
     // Live counters — acquire mutex.
     int getQueueDepth()   const;
@@ -138,7 +144,8 @@ class AsyncRequestManager {
         AsyncRequest()
             : status(REQUEST_QUEUED), requestType(REQUEST_TYPE_SCRIPT)
             , scriptExecuted(false), progress(0.0)
-            , submittedAt(0), startedAt(0), completedAt(0), cancelRequested(0)
+            , submittedAt(0), startedAt(0), completedAt(0), reportOffset(0)
+            , logTotal(0), cancelRequested(0)
         {}
 
         QString       id;
@@ -146,6 +153,7 @@ class AsyncRequestManager {
         RequestType   requestType;
         QString       scriptText;
         QString       scriptFile;
+        QString       reportFile;
         QVariantMap   args;
         QVariant      scriptResult;
         QStringList   outputLines;
@@ -155,11 +163,22 @@ class AsyncRequestManager {
         qint64        submittedAt;
         qint64        startedAt;
         qint64        completedAt;
+        qint64        reportOffset;
+        QByteArray    reportRemainder;
+        QVariantMap   progressDetail;
+        QVariantList  logTail;
+        QVariantList  outputManifest;
+        int           logTotal;
         // Always read/written while holding m_mutex.
         int           cancelRequested;
     };
 
     std::string statusToString(RequestStatus s) const;
+    void ingestReportFromWorkerLocked(AsyncRequest& req);
+    void ingestReportLocked(AsyncRequest& req, bool final = false);
+    void applyReportEventLocked(AsyncRequest& req, const QVariantMap& event);
+    void appendLogLocked(AsyncRequest& req, const QVariantMap& logEntry);
+    std::string observationJson(const AsyncRequest& req) const;
 
     QObject* m_notifyTarget; // DzScriptServerPane*
 
diff --git a/include/DzScriptServerPane.h b/include/DzScriptServerPane.h
index e783100..3942a55 100644
--- a/include/DzScriptServerPane.h
+++ b/include/DzScriptServerPane.h
@@ -161,14 +161,14 @@ public slots:
 	// Qt value types here are protected by AsyncRequestManager's mutex and are
 	// only called directly from documented reentrant Studio 6 worker paths.
 	QString              enqueueAsyncRequest(const QString& scriptText, const QString& scriptFile,
-	                                         const QVariantMap& args,
+	                                         const QString& reportFile, const QVariantMap& args,
 	                                         const QString& idPrefix, qint64& outSubmittedAt,
 	                                         QString& outError);
-	std::pair<int, std::string> getAsyncStatusJson(const std::string& requestId) const;
+	std::pair<int, std::string> getAsyncStatusJson(const std::string& requestId);
 	std::pair<int, std::string> getAsyncResultJson(const std::string& requestId, bool doWait, int timeoutSec);
 	std::pair<int, std::string> cancelAsyncRequestJson(const std::string& requestId, const std::string& clientIP);
 	std::pair<int, std::string> cancelRenderRequestJson(const std::string& requestId, const std::string& clientIP);
-	std::string                 listAsyncRequestsJson(const std::string& statusFilter) const;
+	std::string                 listAsyncRequestsJson(const std::string& statusFilter);
 
 private slots:
 	void onStartClicked();
diff --git a/openapi.yaml b/openapi.yaml
index a92b942..b7daa15 100644
--- a/openapi.yaml
+++ b/openapi.yaml
@@ -281,7 +281,7 @@ paths:
         content:
           application/json:
             schema:
-              $ref: '#/components/schemas/ExecuteRequest'
+              $ref: '#/components/schemas/AsyncExecuteRequest'
       responses:
         '200':
           description: Request queued — poll `/requests/{id}/status` for progress
@@ -315,7 +315,7 @@ paths:
         content:
           application/json:
             schema:
-              $ref: '#/components/schemas/ScriptExecuteRequest'
+              $ref: '#/components/schemas/AsyncScriptExecuteRequest'
       responses:
         '200':
           description: Request queued
@@ -835,6 +835,32 @@ components:
           type: object
           additionalProperties: true
 
+    AsyncExecuteRequest:
+      allOf:
+        - $ref: '#/components/schemas/ExecuteRequest'
+        - type: object
+          properties:
+            reportFile:
+              type: string
+              description: |
+                Optional absolute path to a per-job JSONL event file. The
+                server truncates it at submission and ingests progress, log,
+                output, and manifest events into the async request's
+                observation record. The path is also injected into args as
+                reserved field `__dssReportFile`.
+                DAZ Studio 6 exposes report events while a job is running;
+                Studio 4 parses the final report after execution returns.
+              example: "/path/to/run/job.jsonl"
+
+    AsyncScriptExecuteRequest:
+      allOf:
+        - $ref: '#/components/schemas/ScriptExecuteRequest'
+        - type: object
+          properties:
+            reportFile:
+              type: string
+              description: Optional absolute path to the per-job JSONL event file
+
     DeleteScriptResponse:
       type: object
       required: [success, id]
@@ -902,6 +928,8 @@ components:
           format: float
           minimum: 0
           maximum: 1
+        observation:
+          $ref: '#/components/schemas/JobObservation'
         elapsed_ms:
           type: integer
         queue_position:
@@ -921,6 +949,87 @@ components:
               format: date-time
             status:
               $ref: '#/components/schemas/RequestStatus'
+            observation:
+              $ref: '#/components/schemas/JobObservation'
+
+    JobObservation:
+      type: object
+      required: [progress, log_tail, log_total, log_truncated, output_manifest]
+      properties:
+        progress:
+          nullable: true
+          allOf:
+            - $ref: '#/components/schemas/JobProgress'
+        log_tail:
+          type: array
+          maxItems: 100
+          items:
+            $ref: '#/components/schemas/JobLogEntry'
+        log_total:
+          type: integer
+        log_truncated:
+          type: boolean
+        output_manifest:
+          $ref: '#/components/schemas/JobOutputManifest'
+        report_file:
+          type: string
+
+    JobProgress:
+      type: object
+      required: [value]
+      properties:
+        value:
+          type: number
+          minimum: 0
+          maximum: 1
+        current:
+          type: number
+        total:
+          type: number
+        phase:
+          type: string
+        message:
+          type: string
+        at:
+          type: string
+
+    JobLogEntry:
+      type: object
+      required: [message, level, source]
+      properties:
+        message:
+          type: string
+        level:
+          type: string
+        source:
+          type: string
+        at:
+          type: string
+
+    JobOutputManifest:
+      type: object
+      required: [count, outputs]
+      properties:
+        count:
+          type: integer
+        outputs:
+          type: array
+          items:
+            $ref: '#/components/schemas/JobOutput'
+
+    JobOutput:
+      type: object
+      required: [path]
+      properties:
+        path:
+          type: string
+        kind:
+          type: string
+        label:
+          type: string
+        metadata:
+          type: object
+          additionalProperties: true
 
     CancelResponse:
       type: object
diff --git a/src/AsyncRequestManager.cpp b/src/AsyncRequestManager.cpp
index 1a99e67..3504f96 100644
--- a/src/AsyncRequestManager.cpp
+++ b/src/AsyncRequestManager.cpp
@@ -1,6 +1,10 @@
 #include "AsyncRequestManager.h"
 #include "JsonStd.h"
 #include "MetricsCollector.h"
+#include <QtCore/qdir.h>
+#include <QtCore/qfile.h>
+#include <QtCore/qfileinfo.h>
+#include <QtCore/qobject.h>
 #include <QtCore/qthread.h>
 #include <QtCore/qmetaobject.h>
 
@@ -9,6 +13,13 @@ namespace {
 struct SleepThread : public QThread {
     static void msleep(unsigned long ms) { QThread::msleep(ms); }
 };
+
+QString comparablePath(const QString& path)
+{
+    QFileInfo info(path);
+    const QString canonical = info.canonicalFilePath();
+    return QDir::cleanPath(canonical.isEmpty() ? info.absoluteFilePath() : canonical);
+}
 }
 
 AsyncRequestManager::AsyncRequestManager(QObject* notifyTarget)
@@ -21,7 +32,7 @@ AsyncRequestManager::AsyncRequestManager(QObject* notifyTarget)
 
 AsyncRequestManager::SubmitResult AsyncRequestManager::submit(
     const QString& scriptText, const QString& scriptFile,
-    const QVariantMap& args, const QString& idPrefix)
+    const QString& reportFile, const QVariantMap& args, const QString& idPrefix)
 {
     SubmitResult r;
     r.accepted    = false;
@@ -42,11 +53,32 @@ AsyncRequestManager::SubmitResult AsyncRequestManager::submit(
             return r;
         }
 
+        // The report file is an explicit per-job event stream. Own its
+        // lifecycle from submission so stale events from an earlier run can
+        // never appear in this request's observation record.
+        if (!reportFile.isEmpty()) {
+            if (!scriptFile.isEmpty() &&
+                comparablePath(reportFile).compare(
+                    comparablePath(scriptFile), Qt::CaseInsensitive) == 0) {
+                r.error = "reportFile must not be the scriptFile";
+                return r;
+            }
+            QFile report(reportFile);
+            if (!report.open(QIODevice::WriteOnly | QIODevice::Truncate)) {
+                r.error = QString("Cannot initialize reportFile: %1").arg(reportFile);
+                return r;
+            }
+            report.close();
+        }
+
         AsyncRequest req;
         req.id          = MetricsCollector::generateAsyncId(idPrefix);
         req.scriptText  = scriptText;
         req.scriptFile  = scriptFile;
+        req.reportFile  = reportFile;
         req.args        = args;
+        if (!reportFile.isEmpty())
+            req.args.insert("__dssReportFile", reportFile);
         req.submittedAt = QDateTime::currentMSecsSinceEpoch();
 
         m_requests.insert(req.id, req);
@@ -103,14 +135,211 @@ AsyncRequestManager::SubmitResult AsyncRequestManager::submitRender(
     return r;
 }
 
-std::pair<int, std::string> AsyncRequestManager::getStatusJson(const std::string& requestId) const
+void AsyncRequestManager::appendLogLocked(AsyncRequest& req, const QVariantMap& logEntry)
+{
+    req.logTail.append(logEntry);
+    ++req.logTotal;
+    while (req.logTail.size() > DEFAULT_LOG_TAIL_LINES)
+        req.logTail.removeFirst();
+}
+
+void AsyncRequestManager::applyReportEventLocked(AsyncRequest& req, const QVariantMap& event)
+{
+    const QString type = event.value("type").toString().trimmed().toLower();
+
+    if (type == "progress") {
+        bool valueOk = false;
+        double value = event.value("value").toDouble(&valueOk);
+        if (!valueOk) {
+            bool currentOk = false, totalOk = false;
+            double current = event.value("current").toDouble(&currentOk);
+            double total = event.value("total").toDouble(&totalOk);
+            if (currentOk && totalOk && total > 0.0) {
+                value = current / total;
+                valueOk = true;
+            }
+        }
+        if (!valueOk) return;
+        if (value < 0.0) value = 0.0;
+        if (value > 1.0) value = 1.0;
+
+        QVariantMap detail;
+        detail.insert("value", value);
+        if (event.contains("current")) detail.insert("current", event.value("current"));
+        if (event.contains("total"))   detail.insert("total", event.value("total"));
+        if (event.contains("phase"))   detail.insert("phase", event.value("phase"));
+        if (event.contains("message")) detail.insert("message", event.value("message"));
+        if (event.contains("at"))      detail.insert("at", event.value("at"));
+        req.progress = value;
+        req.progressDetail = detail;
+        return;
+    }
+
+    if (type == "log") {
+        QString message = event.value("message").toString();
+        if (message.isEmpty()) return;
+        QVariantMap entry;
+        entry.insert("message", message);
+        entry.insert("level", event.value("level", "info"));
+        entry.insert("source", event.value("source", "script"));
+        if (event.contains("at")) entry.insert("at", event.value("at"));
+        appendLogLocked(req, entry);
+        return;
+    }
+
+    if (type == "output") {
+        QString path = event.value("path").toString();
+        if (path.isEmpty()) return;
+        QVariantMap output;
+        output.insert("path", path);
+        if (event.contains("kind"))     output.insert("kind", event.value("kind"));
+        if (event.contains("label"))    output.insert("label", event.value("label"));
+        if (event.contains("metadata")) output.insert("metadata", event.value("metadata"));
+
+        for (int i = 0; i < req.outputManifest.size(); ++i) {
+            if (req.outputManifest[i].toMap().value("path").toString() == path) {
+                req.outputManifest[i] = output;
+                return;
+            }
+        }
+        req.outputManifest.append(output);
+        return;
+    }
+
+    if (type == "manifest") {
+        QVariantList outputs = event.value("outputs").toList();
+        for (int i = 0; i < outputs.size(); ++i) {
+            QVariantMap output = outputs[i].toMap();
+            if (!output.isEmpty()) {
+                output.insert("type", "output");
+                applyReportEventLocked(req, output);
+            }
+        }
+    }
+}
+
+void AsyncRequestManager::ingestReportLocked(AsyncRequest& req, bool final)
+{
+#if DAZ_SDK_MAJOR_VERSION < 6
+    // Qt 4's JsonStd::parseObject() constructs a QScriptEngine. This method is
+    // therefore main-thread-only on Studio 4; worker-facing callers go through
+    // ingestReportFromWorkerLocked(), which deliberately no-ops there.
+    Q_ASSERT(!m_notifyTarget || QThread::currentThread() == m_notifyTarget->thread());
+#endif
+    if (req.reportFile.isEmpty()) return;
+
+    QFile report(req.reportFile);
+    if (!report.open(QIODevice::ReadOnly)) return;
+
+    if (report.size() < req.reportOffset) {
+        req.reportOffset = 0;
+        req.reportRemainder.clear();
+        req.progress = 0.0;
+        req.progressDetail.clear();
+        req.logTail.clear();
+        req.logTotal = 0;
+        req.outputManifest.clear();
+    }
+    if (!report.seek(req.reportOffset)) return;
+
+    while (!report.atEnd()) {
+        QByteArray chunk = report.read(64 * 1024);
+        if (chunk.isEmpty()) break;
+        req.reportOffset += chunk.size();
+        req.reportRemainder.append(chunk);
+
+        int newline = -1;
+        while ((newline = req.reportRemainder.indexOf('\n')) >= 0) {
+            QByteArray line = req.reportRemainder.left(newline).trimmed();
+            req.reportRemainder.remove(0, newline + 1);
+            if (line.isEmpty()) continue;
+
+            QVariantMap event;
+            std::string parseError;
+            if (JsonStd::parseObject(line, event, parseError)) {
+                applyReportEventLocked(req, event);
+            } else {
+                QVariantMap warning;
+                warning.insert("level", "warning");
+                warning.insert("source", "server");
+                warning.insert("message", "Ignored malformed job report event");
+                appendLogLocked(req, warning);
+            }
+        }
+
+        if (req.reportRemainder.size() > 64 * 1024) {
+            req.reportRemainder.clear();
+            QVariantMap warning;
+            warning.insert("level", "warning");
+            warning.insert("source", "server");
+            warning.insert("message", "Ignored oversized job report event");
+            appendLogLocked(req, warning);
+        }
+    }
+
+    // JSONL writers should terminate every event with a newline, but accept a
+    // complete final object without one once the script can no longer append.
+    if (final && !req.reportRemainder.trimmed().isEmpty()) {
+        QVariantMap event;
+        std::string parseError;
+        if (JsonStd::parseObject(req.reportRemainder.trimmed(), event, parseError)) {
+            applyReportEventLocked(req, event);
+        } else {
+            QVariantMap warning;
+            warning.insert("level", "warning");
+            warning.insert("source", "server");
+            warning.insert("message", "Ignored malformed final job report event");
+            appendLogLocked(req, warning);
+        }
+        req.reportRemainder.clear();
+    }
+}
+
+void AsyncRequestManager::ingestReportFromWorkerLocked(AsyncRequest& req)
+{
+#if DAZ_SDK_MAJOR_VERSION >= 6
+    ingestReportLocked(req);
+#else
+    // Studio 4 cannot parse report JSON on a raw httplib worker because its
+    // JsonStd fallback uses QScriptEngine. markCompleted() performs the final
+    // ingestion on Daz's main thread after script execution returns.
+    (void)req;
+#endif
+}
+
+std::string AsyncRequestManager::observationJson(const AsyncRequest& req) const
+{
+    QVariantMap manifest;
+    manifest.insert("count", req.outputManifest.size());
+    manifest.insert("outputs", req.outputManifest);
+
+    std::string s = "{\"progress\":";
+    s += req.progressDetail.isEmpty()
+        ? "null"
+        : JsonStd::variantToJson(QVariant(req.progressDetail));
+    s += ",\"log_tail\":" + JsonStd::variantToJson(QVariant(req.logTail));
+    s += ",\"log_total\":" + std::to_string(req.logTotal);
+    s += ",\"log_truncated\":";
+    s += (req.logTotal > req.logTail.size()) ? "true" : "false";
+    s += ",\"output_manifest\":" + JsonStd::variantToJson(QVariant(manifest));
+    if (!req.reportFile.isEmpty()) {
+        s += ",\"report_file\":\"";
+        s += JsonStd::escape(JsonStd::qstrToStd(req.reportFile));
+        s += "\"";
+    }
+    s += "}";
+    return s;
+}
+
+std::pair<int, std::string> AsyncRequestManager::getStatusJson(const std::string& requestId)
 {
     QString qid = QString::fromStdString(requestId);
     QMutexLocker locker(&m_mutex);
     if (!m_requests.contains(qid))
         return {404, "{\"success\":false,\"error\":\"Request not found\"}"};
 
-    const AsyncRequest& req = m_requests.value(qid);
+    AsyncRequest& req = m_requests[qid];
+    ingestReportFromWorkerLocked(req);
     std::string status = statusToString(req.status);
 
     char progBuf[32];
@@ -120,6 +349,7 @@ std::pair<int, std::string> AsyncRequestManager::getStatusJson(const std::string
     s += JsonStd::escape(JsonStd::qstrToStd(req.id));
     s += "\",\"status\":\"" + status + "\"";
     s += ",\"progress\":" + std::string(progBuf);
+    s += ",\"observation\":" + observationJson(req);
 
     if (req.status == REQUEST_RUNNING && req.startedAt > 0) {
         long long elapsed = (long long)(QDateTime::currentMSecsSinceEpoch() - req.startedAt);
@@ -163,42 +393,46 @@ std::pair<int, std::string> AsyncRequestManager::getResultJson(
     if (!m_requests.contains(qid))
         return {404, "{\"success\":false,\"error\":\"Request not found\"}"};
 
-    const AsyncRequest& req = m_requests.value(qid);
-    std::string status = statusToString(req.status);
+    AsyncRequest& mutableReq = m_requests[qid];
+    ingestReportFromWorkerLocked(mutableReq);
+    const AsyncRequest& observedReq = mutableReq;
+    std::string status = statusToString(observedReq.status);
 
     std::string s = "{\"request_id\":\"";
-    s += JsonStd::escape(JsonStd::qstrToStd(req.id));
+    s += JsonStd::escape(JsonStd::qstrToStd(observedReq.id));
     s += "\",\"status\":\"" + status + "\"";
 
-    if (req.status == REQUEST_COMPLETED) {
+    if (observedReq.status == REQUEST_COMPLETED) {
         s += ",\"success\":true";
-        s += ",\"result\":" + JsonStd::variantToJson(req.scriptResult);
+        s += ",\"result\":" + JsonStd::variantToJson(observedReq.scriptResult);
         s += ",\"output\":[";
-        for (int i = 0; i < req.outputLines.size(); ++i) {
+        for (int i = 0; i < observedReq.outputLines.size(); ++i) {
             if (i > 0) s += ",";
-            s += "\"" + JsonStd::escape(JsonStd::qstrToStd(req.outputLines[i])) + "\"";
+            s += "\"" + JsonStd::escape(JsonStd::qstrToStd(observedReq.outputLines[i])) + "\"";
         }
         s += "],\"error\":null";
-    } else if (req.status == REQUEST_FAILED) {
+    } else if (observedReq.status == REQUEST_FAILED) {
         s += ",\"success\":false,\"result\":null,\"output\":[]";
-        s += ",\"error\":\"" + JsonStd::escape(JsonStd::qstrToStd(req.error)) + "\"";
-    } else if (req.status == REQUEST_CANCELLED) {
+        s += ",\"error\":\"" + JsonStd::escape(JsonStd::qstrToStd(observedReq.error)) + "\"";
+    } else if (observedReq.status == REQUEST_CANCELLED) {
         s += ",\"success\":false,\"result\":null,\"output\":[]";
         s += ",\"error\":\"Cancelled\"";
     } else {
         char progBuf[32];
-        std::snprintf(progBuf, sizeof(progBuf), "%.15g", req.progress);
+        std::snprintf(progBuf, sizeof(progBuf), "%.15g", observedReq.progress);
         s += ",\"progress\":";
         s += progBuf;
-        if (req.startedAt > 0) {
-            long long elapsed = (long long)(QDateTime::currentMSecsSinceEpoch() - req.startedAt);
+        if (observedReq.startedAt > 0) {
+            long long elapsed = (long long)(QDateTime::currentMSecsSinceEpoch() - observedReq.startedAt);
             s += ",\"elapsed_ms\":" + std::to_string(elapsed);
         }
     }
 
-    if (req.completedAt > 0 && req.startedAt > 0) {
-        s += ",\"duration_ms\":" + std::to_string((long long)(req.completedAt - req.startedAt));
-        s += ",\"completed_at\":\"" + JsonStd::msecToIso((long long)req.completedAt) + "\"";
+    s += ",\"observation\":" + observationJson(observedReq);
+
+    if (observedReq.completedAt > 0 && observedReq.startedAt > 0) {
+        s += ",\"duration_ms\":" + std::to_string((long long)(observedReq.completedAt - observedReq.startedAt));
+        s += ",\"completed_at\":\"" + JsonStd::msecToIso((long long)observedReq.completedAt) + "\"";
     }
     s += "}";
     return {200, s};
@@ -329,16 +563,17 @@ std::pair<int, std::string> AsyncRequestManager::cancelRenderJson(
     return {200, s};
 }
 
-std::string AsyncRequestManager::listJson(const std::string& statusFilter) const
+std::string AsyncRequestManager::listJson(const std::string& statusFilter)
 {
     QMutexLocker locker(&m_mutex);
 
     int nQueued = 0, nRunning = 0, nCompleted = 0, nFailed = 0, nCancelled = 0;
     std::string items;
 
-    for (QMap<QString, AsyncRequest>::const_iterator it = m_requests.constBegin();
-         it != m_requests.constEnd(); ++it) {
-        const AsyncRequest& req = it.value();
+    for (QMap<QString, AsyncRequest>::iterator it = m_requests.begin();
+         it != m_requests.end(); ++it) {
+        AsyncRequest& req = it.value();
+        ingestReportFromWorkerLocked(req);
         std::string statusStr = statusToString(req.status);
 
         if (!statusFilter.empty() && statusStr != statusFilter)
@@ -351,6 +586,16 @@ std::string AsyncRequestManager::listJson(const std::string& statusFilter) const
         items += "{\"request_id\":\"" + JsonStd::escape(JsonStd::qstrToStd(req.id)) + "\"";
         items += ",\"status\":\"" + statusStr + "\"";
         items += ",\"progress\":" + std::string(progBuf);
+        items += ",\"observation_summary\":{";
+        items += "\"log_total\":" + std::to_string(req.logTotal);
+        items += ",\"log_truncated\":";
+        items += (req.logTotal > req.logTail.size()) ? "true" : "false";
+        items += ",\"output_count\":" + std::to_string(req.outputManifest.size());
+        items += ",\"progress\":";
+        items += req.progressDetail.isEmpty()
+            ? "null"
+            : JsonStd::variantToJson(QVariant(req.progressDetail));
+        items += "}";
         items += ",\"submitted_at\":\"" + JsonStd::msecToIso((long long)req.submittedAt) + "\"}";
 
         switch (req.status) {
@@ -436,6 +681,13 @@ void AsyncRequestManager::markCompleted(const QString& id, bool executed,
 
     AsyncRequest& req  = m_requests[id];
 
+    // This is the main-thread ingestion path for every SDK. It is also the
+    // only report parser used by Studio 4, where JsonStd relies on the
+    // thread-affine QScriptEngine. Parse before the terminal-state guard so a
+    // script that returned after client cancellation or stale-job failure can
+    // still publish its final observation safely.
+    ingestReportLocked(req, true);
+
     // If failStaleRunning() already timed this request out (RUNNING -> FAILED)
     // while the underlying DazScript call was blocked, that terminal state
     // sticks -- a late, real completion must not flip it back to
@@ -444,7 +696,16 @@ void AsyncRequestManager::markCompleted(const QString& id, bool executed,
 
     req.completedAt    = QDateTime::currentMSecsSinceEpoch();
     req.progress       = 1.0;
+    if (executed && !req.cancelRequested && !req.progressDetail.isEmpty())
+        req.progressDetail.insert("value", 1.0);
     req.outputLines    = output;
+    for (int i = 0; i < output.size(); ++i) {
+        QVariantMap entry;
+        entry.insert("level", "info");
+        entry.insert("source", "script");
+        entry.insert("message", output[i]);
+        appendLogLocked(req, entry);
+    }
 
     if (req.cancelRequested) {
         req.status       = REQUEST_CANCELLED;
diff --git a/src/DzScriptServerPane.cpp b/src/DzScriptServerPane.cpp
index 71ba207..86752fa 100644
--- a/src/DzScriptServerPane.cpp
+++ b/src/DzScriptServerPane.cpp
@@ -1465,13 +1465,14 @@ bool DzScriptServerPane::lookupRegistryScript(const std::string& id, std::string
 
 QString DzScriptServerPane::enqueueAsyncRequest(const QString& scriptText,
                                                 const QString& scriptFile,
+                                                const QString& reportFile,
                                                 const QVariantMap& args,
                                                 const QString& idPrefix,
                                                 qint64& outSubmittedAt,
                                                 QString& outError)
 {
 	AsyncRequestManager::SubmitResult r = m_pAsyncMgr->submit(
-		scriptText, scriptFile, args, idPrefix);
+		scriptText, scriptFile, reportFile, args, idPrefix);
 	outSubmittedAt = r.submittedAt;
 	outError       = r.error;
 	if (!r.accepted) {
@@ -1482,7 +1483,7 @@ QString DzScriptServerPane::enqueueAsyncRequest(const QString& scriptText,
 	return r.id;
 }
 
-std::pair<int, std::string> DzScriptServerPane::getAsyncStatusJson(const std::string& requestId) const
+std::pair<int, std::string> DzScriptServerPane::getAsyncStatusJson(const std::string& requestId)
 {
 	return m_pAsyncMgr->getStatusJson(requestId);
 }
@@ -1517,7 +1518,7 @@ std::pair<int, std::string> DzScriptServerPane::cancelRenderRequestJson(const st
 	return result;
 }
 
-std::string DzScriptServerPane::listAsyncRequestsJson(const std::string& statusFilter) const
+std::string DzScriptServerPane::listAsyncRequestsJson(const std::string& statusFilter)
 {
 	return m_pAsyncMgr->listJson(statusFilter);
 }
@@ -1801,7 +1802,8 @@ HttpResult DzScriptServerPane::handleAsyncExecuteEnqueue(
 	qint64  submittedAt  = 0;
 	QString enqueueError;
 	QString requestId = enqueueAsyncRequest(
-		scriptText, scriptFile, body.value("args").toMap(), "execute",
+		scriptText, scriptFile, body.value("reportFile").toString(),
+		body.value("args").toMap(), "execute",
 		submittedAt, enqueueError);
 
 	if (requestId.isEmpty())
@@ -1829,17 +1831,20 @@ HttpResult DzScriptServerPane::handleAsyncScriptEnqueue(
 	QString scriptId   = QString::fromUtf8(scriptIdBytes.constData(), scriptIdBytes.size());
 
 	QVariantMap argsMap;
+	QString reportFile;
 	if (!bodyBytes.isEmpty()) {
 		QVariantMap parsedBody;
 		std::string parseErrDetail;
-		if (JsonStd::parseObject(bodyBytes, parsedBody, parseErrDetail))
+		if (JsonStd::parseObject(bodyBytes, parsedBody, parseErrDetail)) {
 			argsMap = parsedBody.value("args").toMap();
+			reportFile = parsedBody.value("reportFile").toString();
+		}
 	}
 
 	qint64  submittedAt  = 0;
 	QString enqueueError;
 	QString requestId = enqueueAsyncRequest(
-		scriptText, QString(), argsMap, "script", submittedAt, enqueueError);
+		scriptText, QString(), reportFile, argsMap, "script", submittedAt, enqueueError);
 
 	if (requestId.isEmpty())
 		return HttpResult(503, stdToQBA(ErrorResponse::build(
diff --git a/tests/test_api.py b/tests/test_api.py
index b59226f..e4f91b6 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -279,7 +279,7 @@ def test_both_script_and_scriptfile_uses_scriptfile(self):
 
 # ─── Async execution ─────────────────────────────────────────────────────────
 
-def async_execute(script=None, script_file=None, args=None, headers=None):
+def async_execute(script=None, script_file=None, args=None, report_file=None, headers=None):
     """POST /execute/async and return the raw response."""
     payload = {}
     if script is not None:
@@ -288,6 +288,8 @@ def async_execute(script=None, script_file=None, args=None, headers=None):
         payload["scriptFile"] = script_file
     if args is not None:
         payload["args"] = args
+    if report_file is not None:
+        payload["reportFile"] = report_file
     h = auth_headers() if headers is None else headers
     return requests.post(f"{BASE_URL}/execute/async", headers=h, json=payload, timeout=10)
 
@@ -451,6 +453,44 @@ def test_async_inline_after_scriptfile_has_empty_file_identity(self):
             if script_path and os.path.exists(script_path):
                 os.unlink(script_path)
 
+    def test_async_report_file_cannot_overwrite_script_file(self):
+        script_path = ""
+        try:
+            with tempfile.NamedTemporaryFile(
+                mode="w", suffix=".dsa", delete=False, encoding="utf-8"
+            ) as script_file:
+                script_file.write(iife("return 1;"))
+                script_path = os.path.abspath(script_file.name)
+
+            r = async_execute(script_file=script_path, report_file=script_path)
+            self.assertEqual(r.status_code, 503)
+            with open(script_path, encoding="utf-8") as script_file:
+                self.assertIn("return 1", script_file.read())
+        finally:
+            if script_path and os.path.exists(script_path):
+                os.unlink(script_path)
+
+    def test_async_report_file_alias_cannot_overwrite_script_file(self):
+        script_path = ""
+        try:
+            with tempfile.NamedTemporaryFile(
+                mode="w", suffix=".dsa", delete=False, encoding="utf-8"
+            ) as script_file:
+                script_file.write(iife("return 1;"))
+                script_path = os.path.abspath(script_file.name)
+
+            alias_path = os.path.join(
+                os.path.dirname(script_path), ".", os.path.basename(script_path)
+            )
+            self.assertNotEqual(alias_path, script_path)
+            r = async_execute(script_file=script_path, report_file=alias_path)
+            self.assertEqual(r.status_code, 503)
+            with open(script_path, encoding="utf-8") as script_file:
+                self.assertIn("return 1", script_file.read())
+        finally:
+            if script_path and os.path.exists(script_path):
+                os.unlink(script_path)
+
     def test_sync_scriptfile_preserves_file_identity(self):
         script_path = ""
         try:
@@ -598,6 +638,86 @@ def test_async_output_captured(self):
         body = result_r.json()
         self.assertIsInstance(body.get("output"), list)
 
+    def test_async_job_report_tracks_live_progress_logs_and_outputs(self):
+        report_path = ""
+        output_path = os.path.join(tempfile.gettempdir(), "dss-observed-output.png")
+        try:
+            with tempfile.NamedTemporaryFile(
+                mode="w", suffix=".jsonl", delete=False, encoding="utf-8"
+            ) as report:
+                # Submission owns/truncates this file; this stale event must
+                # never leak into the new job record.
+                report.write('{"type":"output","path":"stale.png"}\n')
+                report_path = os.path.abspath(report.name)
+
+            script = iife(
+                "var cfg = getArguments()[0]; "
+                "var events = ''; "
+                "function emit(e) { "
+                "  events += JSON.stringify(e) + '\\n'; "
+                "  var f = new DzFile(cfg.__dssReportFile); "
+                "  if (!f.open(DzFile.WriteOnly)) throw 'report open failed'; "
+                "  f.write(events); f.close(); "
+                "} "
+                "emit({type:'progress', value:0.25, current:1, total:4, "
+                "      phase:'setup', message:'Scene ready'}); "
+                "emit({type:'log', level:'info', message:'started'}); "
+                "var start = Date.now(); while (Date.now() - start < 2500) {} "
+                "for (var i = 0; i < 105; i++) "
+                "  emit({type:'log', level:'debug', message:'line-' + i}); "
+                "emit({type:'output', path:cfg.outputPath, kind:'image', label:'plate'}); "
+                "emit({type:'progress', value:1, current:4, total:4, "
+                "      phase:'done', message:'Complete'}); "
+                "return 'reported';"
+            )
+            r = async_execute(
+                script=script,
+                args={"outputPath": output_path},
+                report_file=report_path,
+            )
+            self.assertEqual(r.status_code, 200)
+            request_id = r.json()["request_id"]
+
+            saw_live_progress = False
+            # A prior cancellation test deliberately leaves a non-render
+            # script finishing on Daz's main thread after its tracker is
+            # already terminal, so this job may remain queued for ~8 seconds.
+            deadline = time.time() + 20
+            while time.time() < deadline:
+                status = requests.get(
+                    f"{BASE_URL}/requests/{request_id}/status",
+                    headers=auth_headers(), timeout=5,
+                ).json()
+                detail = status.get("observation", {}).get("progress") or {}
+                if status.get("status") == "running" and detail.get("value") == 0.25:
+                    saw_live_progress = True
+                    break
+                time.sleep(0.05)
+            self.assertTrue(saw_live_progress, "structured progress was not visible while running")
+
+            body = get_result(request_id, wait=True, timeout=20).json()
+            self.assertEqual(body.get("status"), "completed")
+            observation = body["observation"]
+            self.assertEqual(observation["progress"]["value"], 1)
+            self.assertEqual(observation["progress"]["phase"], "done")
+            self.assertEqual(observation["log_total"], 106)
+            self.assertEqual(len(observation["log_tail"]), 100)
+            self.assertTrue(observation["log_truncated"])
+            manifest = observation["output_manifest"]
+            self.assertEqual(manifest["count"], 1)
+            self.assertEqual(manifest["outputs"][0]["path"], output_path)
+            self.assertNotIn("stale.png", str(manifest))
+
+            listed = requests.get(
+                f"{BASE_URL}/requests", headers=auth_headers(), timeout=5
+            ).json()["requests"]
+            summary = next(x for x in listed if x["request_id"] == request_id)
+            self.assertEqual(summary["observation_summary"]["output_count"], 1)
+            self.assertEqual(summary["observation_summary"]["log_total"], 106)
+        finally:
+            if report_path and os.path.exists(report_path):
+                os.unlink(report_path)
+
 
 # ─── Response shape ───────────────────────────────────────────────────────────
 
diff --git a/tests/test_dazpy.py b/tests/test_dazpy.py
index 2a551ff..8a6e0a1 100644
--- a/tests/test_dazpy.py
+++ b/tests/test_dazpy.py
@@ -11,6 +11,7 @@
 import os
 import sys
 import unittest
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
@@ -76,7 +77,9 @@ def test_posts_script_file_to_async_endpoint_and_returns_request_id(self):
 
         try:
             request_id = client.execute_file_async_submit(
-                "C:/scripts/pose-probe.dsa", args={"mode": "probe"}
+                "C:/scripts/pose-probe.dsa",
+                args={"mode": "probe"},
+                report_file="C:/runs/probe/job.jsonl",
             )
         finally:
             original_session.close()
@@ -87,6 +90,7 @@ def test_posts_script_file_to_async_endpoint_and_returns_request_id(self):
             json={
                 "scriptFile": "C:/scripts/pose-probe.dsa",
                 "args": {"mode": "probe"},
+                "reportFile": "C:/runs/probe/job.jsonl",
             },
             headers={},
             timeout=30.0,
@@ -7496,10 +7500,14 @@ def test_passes_args_through(self):
         client.execute_batch_async(
             [{"body_lines": [], "result_expression": "1"}],
             args={"mode": "probe"},
+            report_file="C:/runs/probe/batch.jsonl",
         )
 
         submitted_payload = client._session.post.call_args[1]["json"]
         self.assertEqual(submitted_payload["args"], {"mode": "probe"})
+        self.assertEqual(
+            submitted_payload["reportFile"], "C:/runs/probe/batch.jsonl"
+        )
 
 
 class TestCallCountBaseline(unittest.TestCase):
@@ -7516,5 +7524,48 @@ def test_batch_add_issues_one_call_for_two_ops(self):
         self.assertEqual(client.execute.call_count, 1)
 
 
+class TestObservationThreadingContract(unittest.TestCase):
+    """Source ratchets for the SDK4/SDK6 report-ingestion boundary.
+
+    The C++ plugin is not built in Python CI, so keep the worker/main-thread
+    split mechanically visible to the gate that runs on every PR.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        source_path = Path(__file__).resolve().parents[1] / "src" / "AsyncRequestManager.cpp"
+        cls.source = source_path.read_text(encoding="utf-8")
+
+    def test_worker_endpoints_use_sdk_guarded_report_ingestion(self):
+        start = self.source.index(
+            "std::pair<int, std::string> AsyncRequestManager::getStatusJson"
+        )
+        end = self.source.index("// ─── Main-thread API", start)
+        worker_api = self.source[start:end]
+
+        self.assertNotIn("ingestReportLocked(", worker_api)
+        self.assertEqual(worker_api.count("ingestReportFromWorkerLocked("), 3)
+
+        helper_start = self.source.index(
+            "void AsyncRequestManager::ingestReportFromWorkerLocked"
+        )
+        helper_end = self.source.index(
+            "std::string AsyncRequestManager::observationJson", helper_start
+        )
+        helper = self.source[helper_start:helper_end]
+        self.assertIn("#if DAZ_SDK_MAJOR_VERSION >= 6", helper)
+        self.assertIn("ingestReportLocked(req);", helper)
+
+    def test_main_thread_completion_ingests_before_terminal_guard(self):
+        start = self.source.index("void AsyncRequestManager::markCompleted")
+        end = self.source.index("void AsyncRequestManager::markCancelled", start)
+        completion = self.source[start:end]
+
+        self.assertLess(
+            completion.index("ingestReportLocked(req, true);"),
+            completion.index("if (req.status != REQUEST_RUNNING) return;"),
+        )
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)
diff --git a/tests/test_dazpy_aio.py b/tests/test_dazpy_aio.py
index 5975fc5..1ac6aca 100644
--- a/tests/test_dazpy_aio.py
+++ b/tests/test_dazpy_aio.py
@@ -79,7 +79,9 @@ async def test_execute_file_async_submit(self):
             json_data={"request_id": "execute-file-123", "status": "queued"}
         )
         request_id = await client.execute_file_async_submit(
-            "C:/scripts/foo.dsa", args={"mode": "probe"}
+            "C:/scripts/foo.dsa",
+            args={"mode": "probe"},
+            report_file="C:/runs/probe/job.jsonl",
         )
         assert request_id == "execute-file-123"
         args, kwargs = mock_http.post.call_args
@@ -87,6 +89,7 @@ async def test_execute_file_async_submit(self):
         assert kwargs["json"] == {
             "scriptFile": "C:/scripts/foo.dsa",
             "args": {"mode": "probe"},
+            "reportFile": "C:/runs/probe/job.jsonl",
         }
 
     @pytest.mark.asyncio
@@ -181,10 +184,12 @@ async def test_passes_args_through(self):
         await client.execute_batch_async(
             [{"body_lines": [], "result_expression": "1"}],
             args={"mode": "probe"},
+            report_file="C:/runs/probe/batch.jsonl",
         )
 
         _, kwargs = mock_http.post.call_args
         assert kwargs["json"]["args"] == {"mode": "probe"}
+        assert kwargs["json"]["reportFile"] == "C:/runs/probe/batch.jsonl"
 
 
 class TestAsyncDazClientRetryOnBusy: