feat(workflows-mcp): add get_all() to ExecutionMemory and flush_memory_on_complete to WorkflowSchema

Author: ibacalu

src/workflows_mcp/engine/execution_memory.py

@@ -99,6 +99,16 @@ async def set(self, key: str, value: str, scope: str = "global") -> None:
         )
         await asyncio.to_thread(self._conn.commit)
 
+    async def get_all(self, scope: str = "global") -> dict[str, str]:
+        """Retrieve all key-value pairs for a given scope."""
+        assert self._conn is not None, "ExecutionMemory not initialized"
+        cursor = await asyncio.to_thread(
+            self._conn.execute,
+            "SELECT key, value FROM execution_context WHERE scope = ? ORDER BY key",
+            (scope,),
+        )
+        return {row[0]: row[1] for row in cursor.fetchall()}
+
     # ------------------------------------------------------------------
     # Conversation Turns
     # ------------------------------------------------------------------

src/workflows_mcp/engine/schema.py

@@ -578,6 +578,10 @@ class WorkflowSchema(BaseModel):
     version: str = Field(default="1.0", pattern=r"^\d+\.\d+(\.\d+)?$")
     author: str | None = Field(default=None)
     tags: list[str] = Field(default_factory=list)
+    flush_memory_on_complete: bool = Field(
+        default=False,
+        description="Flush global-scoped memory entries to durable knowledge on completion",
+    )
 
     # Workflow structure
     inputs: dict[str, WorkflowInputDeclaration] = Field(

tests/test_execution_memory.py

@@ -53,6 +53,27 @@ async def test_scoped_key_missing(self, memory: ExecutionMemory) -> None:
         await memory.set("key", "val", scope="block:a")
         assert await memory.get("key", scope="block:b") is None
 
+    @pytest.mark.asyncio
+    async def test_get_all_returns_global_entries(self, memory: ExecutionMemory) -> None:
+        await memory.set("alpha", "1")
+        await memory.set("beta", "2")
+        result = await memory.get_all("global")
+        assert result == {"alpha": "1", "beta": "2"}
+
+    @pytest.mark.asyncio
+    async def test_get_all_scoped_isolation(self, memory: ExecutionMemory) -> None:
+        """get_all('global') should not include block-scoped entries."""
+        await memory.set("shared", "global_val", scope="global")
+        await memory.set("local", "block_val", scope="block:step_1")
+        result = await memory.get_all("global")
+        assert result == {"shared": "global_val"}
+        assert "local" not in result
+
+    @pytest.mark.asyncio
+    async def test_get_all_empty(self, memory: ExecutionMemory) -> None:
+        result = await memory.get_all("global")
+        assert result == {}
+
 
 # -----------------------------------------------------------------------
 # Conversation Turns