perf(workflow): batch step persistence after execution

flocks/ingest/kafka/manager.py

@@ -91,6 +91,7 @@ def _worker_count_for_trigger(trigger: TriggerDefinition) -> int:
 def _queue_size_for_trigger(trigger: TriggerDefinition) -> int:
     return min(_MAX_QUEUE_SIZE, max(1, int(trigger.concurrency.queueSize)))
 
+
 _KAFKA_STORAGE_LIST_KEYS = DEFAULT_LARGE_LIST_KEYS | frozenset(
     {
         "duplicate_alerts",
@@ -123,6 +124,20 @@ def _strip_execution_only_comments(value: Any) -> Any:
     }
 
 
+def _configured_bool(value: Any, *, default: bool) -> bool:
+    """Parse a boolean trigger input without treating ``"false"`` as true."""
+    if isinstance(value, bool):
+        return value
+    if value is None:
+        return default
+    normalized = str(value).strip().lower()
+    if normalized in {"1", "true", "yes", "on"}:
+        return True
+    if normalized in {"0", "false", "no", "off"}:
+        return False
+    return default
+
+
 def _decode_message(raw: Optional[bytes]) -> Any:
     """Decode a Kafka message value to a Python object.
 
@@ -446,10 +461,13 @@ async def restart_workflow(
             err = "workflow_not_found"
             if startup:
                 self._status[workflow_id] = {"state": "stopped", "error": err}
-                log.info("kafka.workflow_not_found_on_start", {
-                    "workflow_id": workflow_id,
-                    "action": "stale_config_skipped",
-                })
+                log.info(
+                    "kafka.workflow_not_found_on_start",
+                    {
+                        "workflow_id": workflow_id,
+                        "action": "stale_config_skipped",
+                    },
+                )
                 return {"state": "stopped", "error": err}
             self._status[workflow_id] = {"state": "failed", "error": err}
             log.warning("kafka.workflow_not_found", {"workflow_id": workflow_id})
@@ -689,9 +707,7 @@ async def _worker_loop(
         generation_cancel_event: Optional[threading.Event] = None,
     ) -> None:
         run_cancel_event = (
-            generation_cancel_event
-            or self._generation_cancel_events.get(workflow_id)
-            or threading.Event()
+            generation_cancel_event or self._generation_cancel_events.get(workflow_id) or threading.Event()
         )
         while not abort.is_set():
             try:
@@ -748,6 +764,10 @@ async def _trigger_workflow(
         configured_inputs = _strip_execution_only_comments(
             configured_inputs if isinstance(configured_inputs, dict) else {}
         )
+        tool_context_required = _configured_bool(
+            configured_inputs.get("tool_context_required"),
+            default=True,
+        )
         event = build_trigger_event(
             workflow_id=workflow_id,
             trigger=trigger,
@@ -767,15 +787,10 @@ async def _executor(mapped_inputs: Dict[str, Any]) -> Dict[str, Any]:
                 input_params=summarized_inputs,
             )
             exec_id = exec_data["id"]
-            loop = asyncio.get_running_loop()
             start_time = time.time()
             trigger_meta = mapped_inputs.get("_flocks", {}).get("trigger", {})
             trigger_input_keys = list((trigger.mapping or {}).keys()) or [input_key]
             step_recorder = ExecutionStepRecorder(
-                exec_id=exec_id,
-                loop=loop,
-                logger=log,
-                log_event="kafka.execution_step.write_failed",
                 step_compactor=lambda step: _compact_step_for_kafka_storage(
                     step,
                     input_key=input_key,
@@ -784,10 +799,11 @@ async def _executor(mapped_inputs: Dict[str, Any]) -> Dict[str, Any]:
             )
             tool_context = None
             try:
-                tool_context = await build_workflow_tool_context(
-                    workflow_id=workflow_id,
-                    action_name=f"trigger:{trigger.type}",
-                )
+                if tool_context_required:
+                    tool_context = await build_workflow_tool_context(
+                        workflow_id=workflow_id,
+                        action_name=f"trigger:{trigger.type}",
+                    )
                 result = await asyncio.to_thread(
                     run_workflow,
                     workflow=workflow_plan,
@@ -845,9 +861,16 @@ async def _executor(mapped_inputs: Dict[str, Any]) -> Dict[str, Any]:
                     }
                 )
             finally:
-                await cleanup_workflow_tool_context(tool_context)
+                steps = step_recorder.take_steps()
+                if tool_context is not None:
+                    await cleanup_workflow_tool_context(tool_context)
                 try:
-                    await record_execution_result(workflow_id, exec_id, exec_data)
+                    await record_execution_result(
+                        workflow_id,
+                        exec_id,
+                        exec_data,
+                        steps=steps,
+                    )
                 except Exception as exc:
                     log.warning("kafka.exec_record_failed", {"exec_id": exec_id, "error": str(exc)})
             return exec_data

flocks/ingest/syslog/manager.py

@@ -335,10 +335,13 @@ async def restart_workflow(
             err = "workflow_not_found"
             if startup:
                 self._listener_status[workflow_id] = {"state": "stopped", "error": err}
-                log.info("syslog.workflow_not_found_on_start", {
-                    "workflow_id": workflow_id,
-                    "action": "stale_config_skipped",
-                })
+                log.info(
+                    "syslog.workflow_not_found_on_start",
+                    {
+                        "workflow_id": workflow_id,
+                        "action": "stale_config_skipped",
+                    },
+                )
                 return {"state": "stopped", "error": err}
             self._listener_status[workflow_id] = {"state": "failed", "error": err}
             log.warning("syslog.workflow_not_found", {"workflow_id": workflow_id})
@@ -553,9 +556,7 @@ async def _worker_loop(
         of in-flight workflow runs is exactly ``_MAX_CONCURRENT_EXECUTIONS``.
         """
         run_cancel_event = (
-            generation_cancel_event
-            or self._generation_cancel_events.get(workflow_id)
-            or threading.Event()
+            generation_cancel_event or self._generation_cancel_events.get(workflow_id) or threading.Event()
         )
         while not abort.is_set():
             try:
@@ -620,13 +621,7 @@ async def _executor(mapped_inputs: Dict[str, Any]) -> Dict[str, Any]:
                 input_params=summarized_inputs,
             )
             exec_id = exec_data["id"]
-            loop = asyncio.get_running_loop()
-            step_recorder = ExecutionStepRecorder(
-                exec_id=exec_id,
-                loop=loop,
-                logger=log,
-                log_event="syslog.execution_step.write_failed",
-            )
+            step_recorder = ExecutionStepRecorder()
             start_time = time.time()
             trigger_meta = mapped_inputs.get("_flocks", {}).get("trigger", {})
             tool_context = None
@@ -692,9 +687,15 @@ async def _executor(mapped_inputs: Dict[str, Any]) -> Dict[str, Any]:
                     }
                 )
             finally:
+                steps = step_recorder.take_steps()
                 await cleanup_workflow_tool_context(tool_context)
                 try:
-                    await record_execution_result(workflow_id, exec_id, exec_data)
+                    await record_execution_result(
+                        workflow_id,
+                        exec_id,
+                        exec_data,
+                        steps=steps,
+                    )
                 except Exception as exc:
                     log.warning("syslog.exec_record_failed", {"exec_id": exec_id, "error": str(exc)})
             return exec_data