Expose structured observation for async jobs

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

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.

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).
 
 ---

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:

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`."""

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,14 +144,16 @@ 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;
         RequestStatus status;
         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*
 

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();