Vlad/ab 683 (#1190)

* feat: persist outputs to KV

* fix: types

Author: UladzislauK-Writer

src/writer/blueprints.py

@@ -14,6 +14,7 @@
 import writer.blocks.base_block
 import writer.core
 import writer.core_ui
+from writer.journal import JournalRecord
 from writer.ss_types import BlueprintExecutionError, BlueprintExecutionLog, WriterConfigurationError
 
 MAX_DAG_DEPTH = 32
@@ -744,6 +745,12 @@ def run(self) -> Optional[Any]:
                 return self._execute(executor, event)
 
     def _execute(self, executor: ThreadPoolExecutor, abort_event: threading.Event) -> Optional[Any]:
+        journal_record = JournalRecord(
+            execution_environment=self.execution_environment,
+            title=self.status_logger.title,
+            graph=self.graph
+        )
+
         while self.queue or self.futures:
             while self.queue:
                 node: GraphNode = self.queue.pop(0)
@@ -756,6 +763,7 @@ def _execute(self, executor: ThreadPoolExecutor, abort_event: threading.Event) -
                 if abort_event.is_set():
                     self._cancel_all_jobs()
                     self.status_logger.log("Terminated.", entry_type="info", exit="aborted")
+                    journal_record.save(result="stopped")
                     return None
                 else:
                     continue
@@ -768,11 +776,13 @@ def _execute(self, executor: ThreadPoolExecutor, abort_event: threading.Event) -
                 except BlueprintExecutionError as e:
                     self._cancel_all_jobs()
                     self.status_logger.log("Execution failed", entry_type="error", exit=str(e))
+                    journal_record.save(result="error")
                     raise e
                 except BaseException as e:
                     abort_event.set()
                     self._cancel_all_jobs()
                     self.status_logger.log("Execution failed.", entry_type="error", exit=str(e))
+                    journal_record.save(result="error")
                     raise BlueprintExecutionError(
                         f"Blueprint execution was stopped due to an error - {e.__class__.__name__}: {e}"
                     ) from e 
@@ -793,6 +803,7 @@ def _execute(self, executor: ThreadPoolExecutor, abort_event: threading.Event) -
                         self.queue.append(next_node)
 
         self.status_logger.log("Execution completed.", entry_type="info", exit="completed")
+        journal_record.save(result="success")
         return None
 
     def _cancel_local_jobs(self):

src/writer/journal.py

@@ -0,0 +1,72 @@
+import logging
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING, Any, Dict, Literal, Union
+
+from writer.core import Config
+from writer.keyvalue_storage import writer_kv_storage
+
+if TYPE_CHECKING:
+    from writer.blueprints import Graph, GraphNode
+    from writer.core_ui import Component
+
+
+logger = logging.getLogger("journal")
+
+
+class JournalRecord:
+    def __init__(
+        self,
+        execution_environment: Dict,
+        title: str,
+        graph: "Graph"
+    ):
+        self.started_at = datetime.now(timezone.utc)
+        self.instance_type = "editor" if Config.mode == "edit" else "agent"
+
+        self.trigger = {
+            "event": execution_environment.get("context", {}).get("event"),
+            "component": {}
+        }
+
+        component: "Union[GraphNode, Component]"
+        if self.trigger["event"] == "wf-run-blueprint":
+            component = graph.nodes[0].component
+            self.trigger["component"]["type"] = "blueprint"
+            self.trigger["component"]["id"] = graph.nodes[0].component.parentId
+        else:
+            component = graph.get_start_nodes()[0]
+            self.trigger["component"]["type"] = "block"
+            self.trigger["component"]["id"] = graph.get_start_nodes()[0].id
+
+        if "API" in title:
+            if getattr(component, "type", "") == "blueprints_crontrigger":
+                self.trigger["type"] = "Cron"
+            else:
+                self.trigger["type"] = "API"
+        elif "UI" in title:
+            self.trigger["type"] = "UI"
+        else:
+            self.trigger["type"] = "On demand"
+
+        self.graph = graph
+
+    def to_dict(self) -> Dict[str, Any]:
+        block_outputs = {}
+        for graph_node in self.graph.nodes:
+            block_outputs[graph_node.id] = {"result": graph_node.result, "outcome": graph_node.outcome}
+        return {
+            "timestamp": self.started_at.isoformat(),
+            "instanceType": self.instance_type,
+            "trigger": self.trigger,
+            "blockOutputs": block_outputs,
+        }
+
+    def construct_key(self) -> str:
+        return f"wf-journal-{self.instance_type[0]}-{int(self.started_at.timestamp() * 1000)}"
+    
+    def save(self, result: Literal["success", "error", "stopped"]) -> None:
+        if "journal" not in Config.feature_flags or not writer_kv_storage.is_accessible():
+            return
+        data = self.to_dict()
+        data["result"] = result
+        writer_kv_storage.save(self.construct_key(), data)

src/writer/keyvalue_storage.py

@@ -80,4 +80,11 @@ def _request(self, request_func: _WrappedRequestFunc) -> httpx.Response:
         response.raise_for_status()
         return response
 
+    def is_accessible(self) -> bool:
+        if None in self._get_agent_ids():
+            return False
+        if None in (self.api_key, self.api_url):
+            return False
+        return True
+
 writer_kv_storage = KeyValueStorage()

src/writer/logs.py

@@ -246,6 +246,7 @@ def get_logger(level: str = WRITER_LOG_LEVEL):
         "writer": get_logger(),
         "app": get_logger(),
         "from_app": get_logger(),
+        "journal": get_logger(),
         "kv_storage": get_logger(),
         "vault": get_logger(),
         "exec_logger": {