Feat frontend

.gitignore

@@ -10,4 +10,5 @@ test-results/
 .e2e-flaky
 dist/
 *.egg-info/
-.worktrees/
+.worktrees/
+uv.lock

Makefile

@@ -1,17 +1,26 @@
 .DEFAULT_GOAL := help
 
-.PHONY: help test smoke
+.PHONY: help test smoke install-ui build-ui
 
 PYTHON := $(if $(wildcard .venv/bin/python),.venv/bin/python,python3)
+FRONTEND_DIR := agentflow/web/frontend
 
 help:
 	@printf '%s\n' \
  'Available targets:' \
- ' test Run the test suite' \
- ' smoke Run the default smoke pipeline'
+ ' test Run the test suite' \
+ ' smoke Run the default smoke pipeline' \
+ ' install-ui Install frontend dependencies' \
+ ' build-ui Build the frontend dashboard'
 
 test:
 	$(PYTHON) -m pytest -q
 
 smoke:
 	$(PYTHON) -m agentflow run examples/airflow_like.py --output summary
+
+install-ui:
+	cd $(FRONTEND_DIR) && npm install
+
+build-ui:
+	cd $(FRONTEND_DIR) && npm run build

README.md

@@ -22,6 +22,12 @@ python3 -m venv .venv && . .venv/bin/activate
 pip install -e .[dev]
 ```
 
+To build the dashboard (optional, requires Node.js):
+```bash
+make install-ui
+make build-ui
+```
+
 ## Quick Start
 
 ```python

agentflow/app.py

@@ -84,11 +84,28 @@ def create_app(*, store: RunStore | None = None, orchestrator: Orchestrator | No
  app.state.orchestrator = orchestrator
 
  base_dir = os.path.join(os.path.dirname(__file__), "web")
+ frontend_dist = os.path.join(base_dir, "frontend", "dist")
+
+ # Mount Vite static assets
+ if os.path.isdir(frontend_dist):
+ assets_dir = os.path.join(frontend_dist, "assets")
+ if os.path.isdir(assets_dir):
+ app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+ # Public folder/static files inside dist
+ app.mount("/public", StaticFiles(directory=frontend_dist), name="public-dist")
+
+ # Legacy mounts and templates
  templates = Jinja2Templates(directory=os.path.join(base_dir, "templates"))
  app.mount("/static", StaticFiles(directory=os.path.join(base_dir, "static")), name="static")
 
  @app.get("/", response_class=HTMLResponse)
  async def index(request: Request) -> HTMLResponse:
+ react_index = os.path.join(frontend_dist, "index.html")
+ if os.path.isfile(react_index):
+ with open(react_index, "r") as f:
+ return HTMLResponse(content=f.read(), status_code=200)
+
  return templates.TemplateResponse(
  "index.html",
  {"request": request, "example": _load_default_web_example(), "base_dir": os.getcwd()},
@@ -153,6 +170,17 @@ async def get_artifact(run_id: str, node_id: str, name: str) -> PlainTextRespons
  raise HTTPException(status_code=404, detail="artifact not found") from exc
  return PlainTextResponse(content)
 
+ @app.get("/api/runs/{run_id}/scratchboard")
+ async def get_scratchboard(run_id: str) -> PlainTextResponse:
+ from agentflow.scratchboard import SCRATCHBOARD_FILENAME
+ try:
+ path = app.state.store.run_dir(run_id) / SCRATCHBOARD_FILENAME
+ if not path.exists():
+ return PlainTextResponse("")
+ return PlainTextResponse(path.read_text(encoding="utf-8"))
+ except Exception as exc:
+ raise HTTPException(status_code=404, detail="scratchboard not found") from exc
+
  @app.get("/api/runs/{run_id}/stream")
  async def stream_run(run_id: str):
  if run_id not in {run.id for run in app.state.store.list_runs()}:

agentflow/store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import queue
 import threading
 from collections import defaultdict
@@ -12,6 +13,9 @@
 from agentflow.specs import RunEvent, RunRecord
 from agentflow.utils import ensure_dir
 
+# Set up a dedicated logger for sync issues
+sync_logger = logging.getLogger("agentflow.sync")
+
 
 class RunStore:
  def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
@@ -20,24 +24,48 @@ def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
  self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
  self._subscribers: defaultdict[str, set[queue.Queue[RunEvent]]] = defaultdict(set)
  self._events_cache: defaultdict[str, list[RunEvent]] = defaultdict(list)
- self._load_existing_runs()
-
- def _load_existing_runs(self) -> None:
- for run_file in sorted(self.base_dir.glob("*/run.json")):
- run_id = run_file.parent.name
- try:
- run = RunRecord.model_validate_json(run_file.read_text(encoding="utf-8"))
- self._runs[run_id] = run
- events_path = run_file.parent / "events.jsonl"
- if events_path.exists():
- events = [
- RunEvent.model_validate_json(line)
- for line in events_path.read_text(encoding="utf-8").splitlines()
- if line.strip()
- ]
- self._events_cache[run_id] = events
- except (OSError, ValidationError, json.JSONDecodeError, KeyError):
- continue
+ self._mtimes: dict[str, float] = {}
+ self._sync_runs()
+
+ def _sync_runs(self) -> None:
+ """Synchronize in-memory cache with the filesystem."""
+ try:
+ run_files = list(self.base_dir.glob("*/run.json"))
+ for run_file in sorted(run_files):
+ run_id = run_file.parent.name
+ try:
+ mtime = run_file.stat().st_mtime
+ if run_id not in self._runs or self._mtimes.get(run_id, 0) < mtime:
+ content = run_file.read_text(encoding="utf-8")
+ if not content.strip():
+ continue
+
+ run = RunRecord.model_validate_json(content)
+ # Only update if status changed or it's new
+ if run_id not in self._runs or self._runs[run_id].status != run.status:
+ sync_logger.debug(f"Syncing run {run_id}: status {run.status}")
+
+ self._runs[run_id] = run
+ self._mtimes[run_id] = mtime
+
+ # Sync events
+ events_path = run_file.parent / "events.jsonl"
+ if events_path.exists():
+ event_mtime = events_path.stat().st_mtime
+ if self._mtimes.get(f"{run_id}_events", 0) < event_mtime:
+ events = [
+ RunEvent.model_validate_json(line)
+ for line in events_path.read_text(encoding="utf-8").splitlines()
+ if line.strip()
+ ]
+ self._events_cache[run_id] = events
+ self._mtimes[f"{run_id}_events"] = event_mtime
+ except (OSError, ValidationError, json.JSONDecodeError, KeyError) as e:
+ sync_logger.error(f"Failed to sync {run_id}: {e}")
+ continue
+ except OSError as e:
+ sync_logger.error(f"Sync error: {e}")
+ pass
 
  async def create_run(self, record: RunRecord | None = None) -> RunRecord:
  if record is None:
@@ -63,10 +91,26 @@ def cancel_request_path(self, run_id: str) -> Path:
 
  async def persist_run(self, run_id: str) -> None:
  record = self._runs[run_id]
- run_dir = self.run_dir(run_id)
- lock = self._locks[run_id]
- with lock:
- (run_dir / "run.json").write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ path = self.run_dir(run_id) / "run.json"
+
+ # Check if disk is newer before overwriting
+ if path.exists():
+ try:
+ disk_mtime = path.stat().st_mtime
+ if self._mtimes.get(run_id, 0) < disk_mtime:
+ # Disk is newer, reload first to avoid overwriting clinical updates (like failed status)
+ disk_content = path.read_text(encoding="utf-8")
+ disk_record = RunRecord.model_validate_json(disk_content)
+ if disk_record.status in ("failed", "cancelled", "completed"):
+ # Don't overwrite terminal status
+ self._runs[run_id] = disk_record
+ self._mtimes[run_id] = disk_mtime
+ return
+ except Exception as e:
+ sync_logger.error(f"Error checking disk state for {run_id}: {e}")
+
+ path.write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ self._mtimes[run_id] = path.stat().st_mtime
 
  async def append_event(self, run_id: str, event: RunEvent) -> None:
  lock = self._locks[run_id]
@@ -112,12 +156,15 @@ def read_artifact_text(self, run_id: str, node_id: str, name: str) -> str:
  return self.artifact_path(run_id, node_id, name).read_text(encoding="utf-8")
 
  def get_run(self, run_id: str) -> RunRecord:
+ self._sync_runs()
  return self._runs[run_id]
 
  def list_runs(self) -> list[RunRecord]:
+ self._sync_runs()
  return sorted(self._runs.values(), key=lambda run: run.created_at, reverse=True)
 
  def get_events(self, run_id: str) -> list[RunEvent]:
+ self._sync_runs()
  return list(self._events_cache[run_id])
 
  async def subscribe(self, run_id: str) -> queue.Queue[RunEvent]:

agentflow/web/frontend/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

agentflow/web/frontend/README.md

@@ -0,0 +1,73 @@
+# React + TypeScript + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react) uses [Oxc](https://oxc.rs)
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react-swc) uses [SWC](https://swc.rs/)
+
+## React Compiler
+
+The React Compiler is not enabled on this template because of its impact on dev & build performances. To add it, see [this documentation](https://react.dev/learn/react-compiler/installation).
+
+## Expanding the ESLint configuration
+
+If you are developing a production application, we recommend updating the configuration to enable type-aware lint rules:
+
+```js
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+
+ // Remove tseslint.configs.recommended and replace with this
+ tseslint.configs.recommendedTypeChecked,
+ // Alternatively, use this for stricter rules
+ tseslint.configs.strictTypeChecked,
+ // Optionally, add this for stylistic rules
+ tseslint.configs.stylisticTypeChecked,
+
+ // Other configs...
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```
+
+You can also install [eslint-plugin-react-x](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-x) and [eslint-plugin-react-dom](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-dom) for React-specific lint rules:
+
+```js
+// eslint.config.js
+import reactX from 'eslint-plugin-react-x'
+import reactDom from 'eslint-plugin-react-dom'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+ // Enable lint rules for React
+ reactX.configs['recommended-typescript'],
+ // Enable lint rules for React DOM
+ reactDom.configs.recommended,
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```

agentflow/web/frontend/eslint.config.js

@@ -0,0 +1,23 @@
+import js from '@eslint/js'
+import globals from 'globals'
+import reactHooks from 'eslint-plugin-react-hooks'
+import reactRefresh from 'eslint-plugin-react-refresh'
+import tseslint from 'typescript-eslint'
+import { defineConfig, globalIgnores } from 'eslint/config'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ js.configs.recommended,
+ tseslint.configs.recommended,
+ reactHooks.configs.flat.recommended,
+ reactRefresh.configs.vite,
+ ],
+ languageOptions: {
+ ecmaVersion: 2020,
+ globals: globals.browser,
+ },
+ },
+])

agentflow/web/frontend/index.html

@@ -0,0 +1,13 @@
+<!doctype html>
+<htmllang="en">
+<head>
+<metacharset="UTF-8" />
+<linkrel="icon" type="image/svg+xml" href="/favicon.svg" />
+<metaname="viewport" content="width=device-width, initial-scale=1.0" />
+<title>frontend</title>
+</head>
+<body>
+<divid="root"></div>
+<scripttype="module" src="/src/main.tsx"></script>
+</body>
+</html>

.gitignore

@@ -10,4 +10,5 @@ test-results/
 .e2e-flaky
 dist/
 *.egg-info/
-.worktrees/
+.worktrees/
+uv.lock

Makefile

@@ -1,17 +1,26 @@
 .DEFAULT_GOAL := help
 
-.PHONY: help test smoke
+.PHONY: help test smoke install-ui build-ui
 
 PYTHON := $(if $(wildcard .venv/bin/python),.venv/bin/python,python3)
+FRONTEND_DIR := agentflow/web/frontend
 
 help:
 	@printf '%s\n' \
  'Available targets:' \
- ' test Run the test suite' \
- ' smoke Run the default smoke pipeline'
+ ' test Run the test suite' \
+ ' smoke Run the default smoke pipeline' \
+ ' install-ui Install frontend dependencies' \
+ ' build-ui Build the frontend dashboard'
 
 test:
 	$(PYTHON) -m pytest -q
 
 smoke:
 	$(PYTHON) -m agentflow run examples/airflow_like.py --output summary
+
+install-ui:
+	cd $(FRONTEND_DIR) && npm install
+
+build-ui:
+	cd $(FRONTEND_DIR) && npm run build

README.md

@@ -22,6 +22,12 @@ python3 -m venv .venv && . .venv/bin/activate
 pip install -e .[dev]
 ```
 
+To build the dashboard (optional, requires Node.js):
+```bash
+make install-ui
+make build-ui
+```
+
 ## Quick Start
 
 ```python

agentflow/app.py

@@ -84,11 +84,28 @@ def create_app(*, store: RunStore | None = None, orchestrator: Orchestrator | No
  app.state.orchestrator = orchestrator
 
  base_dir = os.path.join(os.path.dirname(__file__), "web")
+ frontend_dist = os.path.join(base_dir, "frontend", "dist")
+
+ # Mount Vite static assets
+ if os.path.isdir(frontend_dist):
+ assets_dir = os.path.join(frontend_dist, "assets")
+ if os.path.isdir(assets_dir):
+ app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+ # Public folder/static files inside dist
+ app.mount("/public", StaticFiles(directory=frontend_dist), name="public-dist")
+
+ # Legacy mounts and templates
  templates = Jinja2Templates(directory=os.path.join(base_dir, "templates"))
  app.mount("/static", StaticFiles(directory=os.path.join(base_dir, "static")), name="static")
 
  @app.get("/", response_class=HTMLResponse)
  async def index(request: Request) -> HTMLResponse:
+ react_index = os.path.join(frontend_dist, "index.html")
+ if os.path.isfile(react_index):
+ with open(react_index, "r") as f:
+ return HTMLResponse(content=f.read(), status_code=200)
+
  return templates.TemplateResponse(
  "index.html",
  {"request": request, "example": _load_default_web_example(), "base_dir": os.getcwd()},
@@ -153,6 +170,17 @@ async def get_artifact(run_id: str, node_id: str, name: str) -> PlainTextRespons
  raise HTTPException(status_code=404, detail="artifact not found") from exc
  return PlainTextResponse(content)
 
+ @app.get("/api/runs/{run_id}/scratchboard")
+ async def get_scratchboard(run_id: str) -> PlainTextResponse:
+ from agentflow.scratchboard import SCRATCHBOARD_FILENAME
+ try:
+ path = app.state.store.run_dir(run_id) / SCRATCHBOARD_FILENAME
+ if not path.exists():
+ return PlainTextResponse("")
+ return PlainTextResponse(path.read_text(encoding="utf-8"))
+ except Exception as exc:
+ raise HTTPException(status_code=404, detail="scratchboard not found") from exc
+
  @app.get("/api/runs/{run_id}/stream")
  async def stream_run(run_id: str):
  if run_id not in {run.id for run in app.state.store.list_runs()}:

agentflow/store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import queue
 import threading
 from collections import defaultdict
@@ -12,6 +13,9 @@
 from agentflow.specs import RunEvent, RunRecord
 from agentflow.utils import ensure_dir
 
+# Set up a dedicated logger for sync issues
+sync_logger = logging.getLogger("agentflow.sync")
+
 
 class RunStore:
  def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
@@ -20,24 +24,48 @@ def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
  self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
  self._subscribers: defaultdict[str, set[queue.Queue[RunEvent]]] = defaultdict(set)
  self._events_cache: defaultdict[str, list[RunEvent]] = defaultdict(list)
- self._load_existing_runs()
-
- def _load_existing_runs(self) -> None:
- for run_file in sorted(self.base_dir.glob("*/run.json")):
- run_id = run_file.parent.name
- try:
- run = RunRecord.model_validate_json(run_file.read_text(encoding="utf-8"))
- self._runs[run_id] = run
- events_path = run_file.parent / "events.jsonl"
- if events_path.exists():
- events = [
- RunEvent.model_validate_json(line)
- for line in events_path.read_text(encoding="utf-8").splitlines()
- if line.strip()
- ]
- self._events_cache[run_id] = events
- except (OSError, ValidationError, json.JSONDecodeError, KeyError):
- continue
+ self._mtimes: dict[str, float] = {}
+ self._sync_runs()
+
+ def _sync_runs(self) -> None:
+ """Synchronize in-memory cache with the filesystem."""
+ try:
+ run_files = list(self.base_dir.glob("*/run.json"))
+ for run_file in sorted(run_files):
+ run_id = run_file.parent.name
+ try:
+ mtime = run_file.stat().st_mtime
+ if run_id not in self._runs or self._mtimes.get(run_id, 0) < mtime:
+ content = run_file.read_text(encoding="utf-8")
+ if not content.strip():
+ continue
+
+ run = RunRecord.model_validate_json(content)
+ # Only update if status changed or it's new
+ if run_id not in self._runs or self._runs[run_id].status != run.status:
+ sync_logger.debug(f"Syncing run {run_id}: status {run.status}")
+
+ self._runs[run_id] = run
+ self._mtimes[run_id] = mtime
+
+ # Sync events
+ events_path = run_file.parent / "events.jsonl"
+ if events_path.exists():
+ event_mtime = events_path.stat().st_mtime
+ if self._mtimes.get(f"{run_id}_events", 0) < event_mtime:
+ events = [
+ RunEvent.model_validate_json(line)
+ for line in events_path.read_text(encoding="utf-8").splitlines()
+ if line.strip()
+ ]
+ self._events_cache[run_id] = events
+ self._mtimes[f"{run_id}_events"] = event_mtime
+ except (OSError, ValidationError, json.JSONDecodeError, KeyError) as e:
+ sync_logger.error(f"Failed to sync {run_id}: {e}")
+ continue
+ except OSError as e:
+ sync_logger.error(f"Sync error: {e}")
+ pass
 
  async def create_run(self, record: RunRecord | None = None) -> RunRecord:
  if record is None:
@@ -63,10 +91,26 @@ def cancel_request_path(self, run_id: str) -> Path:
 
  async def persist_run(self, run_id: str) -> None:
  record = self._runs[run_id]
- run_dir = self.run_dir(run_id)
- lock = self._locks[run_id]
- with lock:
- (run_dir / "run.json").write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ path = self.run_dir(run_id) / "run.json"
+
+ # Check if disk is newer before overwriting
+ if path.exists():
+ try:
+ disk_mtime = path.stat().st_mtime
+ if self._mtimes.get(run_id, 0) < disk_mtime:
+ # Disk is newer, reload first to avoid overwriting clinical updates (like failed status)
+ disk_content = path.read_text(encoding="utf-8")
+ disk_record = RunRecord.model_validate_json(disk_content)
+ if disk_record.status in ("failed", "cancelled", "completed"):
+ # Don't overwrite terminal status
+ self._runs[run_id] = disk_record
+ self._mtimes[run_id] = disk_mtime
+ return
+ except Exception as e:
+ sync_logger.error(f"Error checking disk state for {run_id}: {e}")
+
+ path.write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ self._mtimes[run_id] = path.stat().st_mtime
 
  async def append_event(self, run_id: str, event: RunEvent) -> None:
  lock = self._locks[run_id]
@@ -112,12 +156,15 @@ def read_artifact_text(self, run_id: str, node_id: str, name: str) -> str:
  return self.artifact_path(run_id, node_id, name).read_text(encoding="utf-8")
 
  def get_run(self, run_id: str) -> RunRecord:
+ self._sync_runs()
  return self._runs[run_id]
 
  def list_runs(self) -> list[RunRecord]:
+ self._sync_runs()
  return sorted(self._runs.values(), key=lambda run: run.created_at, reverse=True)
 
  def get_events(self, run_id: str) -> list[RunEvent]:
+ self._sync_runs()
  return list(self._events_cache[run_id])
 
  async def subscribe(self, run_id: str) -> queue.Queue[RunEvent]:

agentflow/web/frontend/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

agentflow/web/frontend/README.md

@@ -0,0 +1,73 @@
+# React + TypeScript + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react) uses [Oxc](https://oxc.rs)
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react-swc) uses [SWC](https://swc.rs/)
+
+## React Compiler
+
+The React Compiler is not enabled on this template because of its impact on dev & build performances. To add it, see [this documentation](https://react.dev/learn/react-compiler/installation).
+
+## Expanding the ESLint configuration
+
+If you are developing a production application, we recommend updating the configuration to enable type-aware lint rules:
+
+```js
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+
+ // Remove tseslint.configs.recommended and replace with this
+ tseslint.configs.recommendedTypeChecked,
+ // Alternatively, use this for stricter rules
+ tseslint.configs.strictTypeChecked,
+ // Optionally, add this for stylistic rules
+ tseslint.configs.stylisticTypeChecked,
+
+ // Other configs...
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```
+
+You can also install [eslint-plugin-react-x](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-x) and [eslint-plugin-react-dom](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-dom) for React-specific lint rules:
+
+```js
+// eslint.config.js
+import reactX from 'eslint-plugin-react-x'
+import reactDom from 'eslint-plugin-react-dom'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+ // Enable lint rules for React
+ reactX.configs['recommended-typescript'],
+ // Enable lint rules for React DOM
+ reactDom.configs.recommended,
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```

agentflow/web/frontend/eslint.config.js

@@ -0,0 +1,23 @@
+import js from '@eslint/js'
+import globals from 'globals'
+import reactHooks from 'eslint-plugin-react-hooks'
+import reactRefresh from 'eslint-plugin-react-refresh'
+import tseslint from 'typescript-eslint'
+import { defineConfig, globalIgnores } from 'eslint/config'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ js.configs.recommended,
+ tseslint.configs.recommended,
+ reactHooks.configs.flat.recommended,
+ reactRefresh.configs.vite,
+ ],
+ languageOptions: {
+ ecmaVersion: 2020,
+ globals: globals.browser,
+ },
+ },
+])

agentflow/web/frontend/index.html

@@ -0,0 +1,13 @@
+<!doctype html>
+<htmllang="en">
+<head>
+<metacharset="UTF-8" />
+<linkrel="icon" type="image/svg+xml" href="/favicon.svg" />
+<metaname="viewport" content="width=device-width, initial-scale=1.0" />
+<title>frontend</title>
+</head>
+<body>
+<divid="root"></div>
+<scripttype="module" src="/src/main.tsx"></script>
+</body>
+</html>

.gitignore

@@ -10,4 +10,5 @@ test-results/
 .e2e-flaky
 dist/
 *.egg-info/
-.worktrees/
+.worktrees/
+uv.lock

Makefile

@@ -1,17 +1,26 @@
 .DEFAULT_GOAL := help
 
-.PHONY: help test smoke
+.PHONY: help test smoke install-ui build-ui
 
 PYTHON := $(if $(wildcard .venv/bin/python),.venv/bin/python,python3)
+FRONTEND_DIR := agentflow/web/frontend
 
 help:
 	@printf '%s\n' \
  'Available targets:' \
- ' test Run the test suite' \
- ' smoke Run the default smoke pipeline'
+ ' test Run the test suite' \
+ ' smoke Run the default smoke pipeline' \
+ ' install-ui Install frontend dependencies' \
+ ' build-ui Build the frontend dashboard'
 
 test:
 	$(PYTHON) -m pytest -q
 
 smoke:
 	$(PYTHON) -m agentflow run examples/airflow_like.py --output summary
+
+install-ui:
+	cd $(FRONTEND_DIR) && npm install
+
+build-ui:
+	cd $(FRONTEND_DIR) && npm run build

README.md

@@ -22,6 +22,12 @@ python3 -m venv .venv && . .venv/bin/activate
 pip install -e .[dev]
 ```
 
+To build the dashboard (optional, requires Node.js):
+```bash
+make install-ui
+make build-ui
+```
+
 ## Quick Start
 
 ```python

agentflow/app.py

@@ -84,11 +84,28 @@ def create_app(*, store: RunStore | None = None, orchestrator: Orchestrator | No
  app.state.orchestrator = orchestrator
 
  base_dir = os.path.join(os.path.dirname(__file__), "web")
+ frontend_dist = os.path.join(base_dir, "frontend", "dist")
+
+ # Mount Vite static assets
+ if os.path.isdir(frontend_dist):
+ assets_dir = os.path.join(frontend_dist, "assets")
+ if os.path.isdir(assets_dir):
+ app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+ # Public folder/static files inside dist
+ app.mount("/public", StaticFiles(directory=frontend_dist), name="public-dist")
+
+ # Legacy mounts and templates
  templates = Jinja2Templates(directory=os.path.join(base_dir, "templates"))
  app.mount("/static", StaticFiles(directory=os.path.join(base_dir, "static")), name="static")
 
  @app.get("/", response_class=HTMLResponse)
  async def index(request: Request) -> HTMLResponse:
+ react_index = os.path.join(frontend_dist, "index.html")
+ if os.path.isfile(react_index):
+ with open(react_index, "r") as f:
+ return HTMLResponse(content=f.read(), status_code=200)
+
  return templates.TemplateResponse(
  "index.html",
  {"request": request, "example": _load_default_web_example(), "base_dir": os.getcwd()},
@@ -153,6 +170,17 @@ async def get_artifact(run_id: str, node_id: str, name: str) -> PlainTextRespons
  raise HTTPException(status_code=404, detail="artifact not found") from exc
  return PlainTextResponse(content)
 
+ @app.get("/api/runs/{run_id}/scratchboard")
+ async def get_scratchboard(run_id: str) -> PlainTextResponse:
+ from agentflow.scratchboard import SCRATCHBOARD_FILENAME
+ try:
+ path = app.state.store.run_dir(run_id) / SCRATCHBOARD_FILENAME
+ if not path.exists():
+ return PlainTextResponse("")
+ return PlainTextResponse(path.read_text(encoding="utf-8"))
+ except Exception as exc:
+ raise HTTPException(status_code=404, detail="scratchboard not found") from exc
+
  @app.get("/api/runs/{run_id}/stream")
  async def stream_run(run_id: str):
  if run_id not in {run.id for run in app.state.store.list_runs()}:

agentflow/store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import queue
 import threading
 from collections import defaultdict
@@ -12,6 +13,9 @@
 from agentflow.specs import RunEvent, RunRecord
 from agentflow.utils import ensure_dir
 
+# Set up a dedicated logger for sync issues
+sync_logger = logging.getLogger("agentflow.sync")
+
 
 class RunStore:
  def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
@@ -20,24 +24,48 @@ def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
  self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
  self._subscribers: defaultdict[str, set[queue.Queue[RunEvent]]] = defaultdict(set)
  self._events_cache: defaultdict[str, list[RunEvent]] = defaultdict(list)
- self._load_existing_runs()
-
- def _load_existing_runs(self) -> None:
- for run_file in sorted(self.base_dir.glob("*/run.json")):
- run_id = run_file.parent.name
- try:
- run = RunRecord.model_validate_json(run_file.read_text(encoding="utf-8"))
- self._runs[run_id] = run
- events_path = run_file.parent / "events.jsonl"
- if events_path.exists():
- events = [
- RunEvent.model_validate_json(line)
- for line in events_path.read_text(encoding="utf-8").splitlines()
- if line.strip()
- ]
- self._events_cache[run_id] = events
- except (OSError, ValidationError, json.JSONDecodeError, KeyError):
- continue
+ self._mtimes: dict[str, float] = {}
+ self._sync_runs()
+
+ def _sync_runs(self) -> None:
+ """Synchronize in-memory cache with the filesystem."""
+ try:
+ run_files = list(self.base_dir.glob("*/run.json"))
+ for run_file in sorted(run_files):
+ run_id = run_file.parent.name
+ try:
+ mtime = run_file.stat().st_mtime
+ if run_id not in self._runs or self._mtimes.get(run_id, 0) < mtime:
+ content = run_file.read_text(encoding="utf-8")
+ if not content.strip():
+ continue
+
+ run = RunRecord.model_validate_json(content)
+ # Only update if status changed or it's new
+ if run_id not in self._runs or self._runs[run_id].status != run.status:
+ sync_logger.debug(f"Syncing run {run_id}: status {run.status}")
+
+ self._runs[run_id] = run
+ self._mtimes[run_id] = mtime
+
+ # Sync events
+ events_path = run_file.parent / "events.jsonl"
+ if events_path.exists():
+ event_mtime = events_path.stat().st_mtime
+ if self._mtimes.get(f"{run_id}_events", 0) < event_mtime:
+ events = [
+ RunEvent.model_validate_json(line)
+ for line in events_path.read_text(encoding="utf-8").splitlines()
+ if line.strip()
+ ]
+ self._events_cache[run_id] = events
+ self._mtimes[f"{run_id}_events"] = event_mtime
+ except (OSError, ValidationError, json.JSONDecodeError, KeyError) as e:
+ sync_logger.error(f"Failed to sync {run_id}: {e}")
+ continue
+ except OSError as e:
+ sync_logger.error(f"Sync error: {e}")
+ pass
 
  async def create_run(self, record: RunRecord | None = None) -> RunRecord:
  if record is None:
@@ -63,10 +91,26 @@ def cancel_request_path(self, run_id: str) -> Path:
 
  async def persist_run(self, run_id: str) -> None:
  record = self._runs[run_id]
- run_dir = self.run_dir(run_id)
- lock = self._locks[run_id]
- with lock:
- (run_dir / "run.json").write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ path = self.run_dir(run_id) / "run.json"
+
+ # Check if disk is newer before overwriting
+ if path.exists():
+ try:
+ disk_mtime = path.stat().st_mtime
+ if self._mtimes.get(run_id, 0) < disk_mtime:
+ # Disk is newer, reload first to avoid overwriting clinical updates (like failed status)
+ disk_content = path.read_text(encoding="utf-8")
+ disk_record = RunRecord.model_validate_json(disk_content)
+ if disk_record.status in ("failed", "cancelled", "completed"):
+ # Don't overwrite terminal status
+ self._runs[run_id] = disk_record
+ self._mtimes[run_id] = disk_mtime
+ return
+ except Exception as e:
+ sync_logger.error(f"Error checking disk state for {run_id}: {e}")
+
+ path.write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ self._mtimes[run_id] = path.stat().st_mtime
 
  async def append_event(self, run_id: str, event: RunEvent) -> None:
  lock = self._locks[run_id]
@@ -112,12 +156,15 @@ def read_artifact_text(self, run_id: str, node_id: str, name: str) -> str:
  return self.artifact_path(run_id, node_id, name).read_text(encoding="utf-8")
 
  def get_run(self, run_id: str) -> RunRecord:
+ self._sync_runs()
  return self._runs[run_id]
 
  def list_runs(self) -> list[RunRecord]:
+ self._sync_runs()
  return sorted(self._runs.values(), key=lambda run: run.created_at, reverse=True)
 
  def get_events(self, run_id: str) -> list[RunEvent]:
+ self._sync_runs()
  return list(self._events_cache[run_id])
 
  async def subscribe(self, run_id: str) -> queue.Queue[RunEvent]:

agentflow/web/frontend/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

agentflow/web/frontend/README.md

@@ -0,0 +1,73 @@
+# React + TypeScript + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react) uses [Oxc](https://oxc.rs)
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react-swc) uses [SWC](https://swc.rs/)
+
+## React Compiler
+
+The React Compiler is not enabled on this template because of its impact on dev & build performances. To add it, see [this documentation](https://react.dev/learn/react-compiler/installation).
+
+## Expanding the ESLint configuration
+
+If you are developing a production application, we recommend updating the configuration to enable type-aware lint rules:
+
+```js
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+
+ // Remove tseslint.configs.recommended and replace with this
+ tseslint.configs.recommendedTypeChecked,
+ // Alternatively, use this for stricter rules
+ tseslint.configs.strictTypeChecked,
+ // Optionally, add this for stylistic rules
+ tseslint.configs.stylisticTypeChecked,
+
+ // Other configs...
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```
+
+You can also install [eslint-plugin-react-x](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-x) and [eslint-plugin-react-dom](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-dom) for React-specific lint rules:
+
+```js
+// eslint.config.js
+import reactX from 'eslint-plugin-react-x'
+import reactDom from 'eslint-plugin-react-dom'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+ // Enable lint rules for React
+ reactX.configs['recommended-typescript'],
+ // Enable lint rules for React DOM
+ reactDom.configs.recommended,
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```

agentflow/web/frontend/eslint.config.js

@@ -0,0 +1,23 @@
+import js from '@eslint/js'
+import globals from 'globals'
+import reactHooks from 'eslint-plugin-react-hooks'
+import reactRefresh from 'eslint-plugin-react-refresh'
+import tseslint from 'typescript-eslint'
+import { defineConfig, globalIgnores } from 'eslint/config'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ js.configs.recommended,
+ tseslint.configs.recommended,
+ reactHooks.configs.flat.recommended,
+ reactRefresh.configs.vite,
+ ],
+ languageOptions: {
+ ecmaVersion: 2020,
+ globals: globals.browser,
+ },
+ },
+])

agentflow/web/frontend/index.html

@@ -0,0 +1,13 @@
+<!doctype html>
+<htmllang="en">
+<head>
+<metacharset="UTF-8" />
+<linkrel="icon" type="image/svg+xml" href="/favicon.svg" />
+<metaname="viewport" content="width=device-width, initial-scale=1.0" />
+<title>frontend</title>
+</head>
+<body>
+<divid="root"></div>
+<scripttype="module" src="/src/main.tsx"></script>
+</body>
+</html>

.gitignore

@@ -10,4 +10,5 @@ test-results/
 .e2e-flaky
 dist/
 *.egg-info/
-.worktrees/
+.worktrees/
+uv.lock

Makefile

@@ -1,17 +1,26 @@
 .DEFAULT_GOAL := help
 
-.PHONY: help test smoke
+.PHONY: help test smoke install-ui build-ui
 
 PYTHON := $(if $(wildcard .venv/bin/python),.venv/bin/python,python3)
+FRONTEND_DIR := agentflow/web/frontend
 
 help:
 	@printf '%s\n' \
  'Available targets:' \
- ' test Run the test suite' \
- ' smoke Run the default smoke pipeline'
+ ' test Run the test suite' \
+ ' smoke Run the default smoke pipeline' \
+ ' install-ui Install frontend dependencies' \
+ ' build-ui Build the frontend dashboard'
 
 test:
 	$(PYTHON) -m pytest -q
 
 smoke:
 	$(PYTHON) -m agentflow run examples/airflow_like.py --output summary
+
+install-ui:
+	cd $(FRONTEND_DIR) && npm install
+
+build-ui:
+	cd $(FRONTEND_DIR) && npm run build

README.md

@@ -22,6 +22,12 @@ python3 -m venv .venv && . .venv/bin/activate
 pip install -e .[dev]
 ```
 
+To build the dashboard (optional, requires Node.js):
+```bash
+make install-ui
+make build-ui
+```
+
 ## Quick Start
 
 ```python

agentflow/app.py

@@ -84,11 +84,28 @@ def create_app(*, store: RunStore | None = None, orchestrator: Orchestrator | No
  app.state.orchestrator = orchestrator
 
  base_dir = os.path.join(os.path.dirname(__file__), "web")
+ frontend_dist = os.path.join(base_dir, "frontend", "dist")
+
+ # Mount Vite static assets
+ if os.path.isdir(frontend_dist):
+ assets_dir = os.path.join(frontend_dist, "assets")
+ if os.path.isdir(assets_dir):
+ app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+ # Public folder/static files inside dist
+ app.mount("/public", StaticFiles(directory=frontend_dist), name="public-dist")
+
+ # Legacy mounts and templates
  templates = Jinja2Templates(directory=os.path.join(base_dir, "templates"))
  app.mount("/static", StaticFiles(directory=os.path.join(base_dir, "static")), name="static")
 
  @app.get("/", response_class=HTMLResponse)
  async def index(request: Request) -> HTMLResponse:
+ react_index = os.path.join(frontend_dist, "index.html")
+ if os.path.isfile(react_index):
+ with open(react_index, "r") as f:
+ return HTMLResponse(content=f.read(), status_code=200)
+
  return templates.TemplateResponse(
  "index.html",
  {"request": request, "example": _load_default_web_example(), "base_dir": os.getcwd()},
@@ -153,6 +170,17 @@ async def get_artifact(run_id: str, node_id: str, name: str) -> PlainTextRespons
  raise HTTPException(status_code=404, detail="artifact not found") from exc
  return PlainTextResponse(content)
 
+ @app.get("/api/runs/{run_id}/scratchboard")
+ async def get_scratchboard(run_id: str) -> PlainTextResponse:
+ from agentflow.scratchboard import SCRATCHBOARD_FILENAME
+ try:
+ path = app.state.store.run_dir(run_id) / SCRATCHBOARD_FILENAME
+ if not path.exists():
+ return PlainTextResponse("")
+ return PlainTextResponse(path.read_text(encoding="utf-8"))
+ except Exception as exc:
+ raise HTTPException(status_code=404, detail="scratchboard not found") from exc
+
  @app.get("/api/runs/{run_id}/stream")
  async def stream_run(run_id: str):
  if run_id not in {run.id for run in app.state.store.list_runs()}:

agentflow/store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import queue
 import threading
 from collections import defaultdict
@@ -12,6 +13,9 @@
 from agentflow.specs import RunEvent, RunRecord
 from agentflow.utils import ensure_dir
 
+# Set up a dedicated logger for sync issues
+sync_logger = logging.getLogger("agentflow.sync")
+
 
 class RunStore:
  def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
@@ -20,24 +24,48 @@ def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
  self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
  self._subscribers: defaultdict[str, set[queue.Queue[RunEvent]]] = defaultdict(set)
  self._events_cache: defaultdict[str, list[RunEvent]] = defaultdict(list)
- self._load_existing_runs()
-
- def _load_existing_runs(self) -> None:
- for run_file in sorted(self.base_dir.glob("*/run.json")):
- run_id = run_file.parent.name
- try:
- run = RunRecord.model_validate_json(run_file.read_text(encoding="utf-8"))
- self._runs[run_id] = run
- events_path = run_file.parent / "events.jsonl"
- if events_path.exists():
- events = [
- RunEvent.model_validate_json(line)
- for line in events_path.read_text(encoding="utf-8").splitlines()
- if line.strip()
- ]
- self._events_cache[run_id] = events
- except (OSError, ValidationError, json.JSONDecodeError, KeyError):
- continue
+ self._mtimes: dict[str, float] = {}
+ self._sync_runs()
+
+ def _sync_runs(self) -> None:
+ """Synchronize in-memory cache with the filesystem."""
+ try:
+ run_files = list(self.base_dir.glob("*/run.json"))
+ for run_file in sorted(run_files):
+ run_id = run_file.parent.name
+ try:
+ mtime = run_file.stat().st_mtime
+ if run_id not in self._runs or self._mtimes.get(run_id, 0) < mtime:
+ content = run_file.read_text(encoding="utf-8")
+ if not content.strip():
+ continue
+
+ run = RunRecord.model_validate_json(content)
+ # Only update if status changed or it's new
+ if run_id not in self._runs or self._runs[run_id].status != run.status:
+ sync_logger.debug(f"Syncing run {run_id}: status {run.status}")
+
+ self._runs[run_id] = run
+ self._mtimes[run_id] = mtime
+
+ # Sync events
+ events_path = run_file.parent / "events.jsonl"
+ if events_path.exists():
+ event_mtime = events_path.stat().st_mtime
+ if self._mtimes.get(f"{run_id}_events", 0) < event_mtime:
+ events = [
+ RunEvent.model_validate_json(line)
+ for line in events_path.read_text(encoding="utf-8").splitlines()
+ if line.strip()
+ ]
+ self._events_cache[run_id] = events
+ self._mtimes[f"{run_id}_events"] = event_mtime
+ except (OSError, ValidationError, json.JSONDecodeError, KeyError) as e:
+ sync_logger.error(f"Failed to sync {run_id}: {e}")
+ continue
+ except OSError as e:
+ sync_logger.error(f"Sync error: {e}")
+ pass
 
  async def create_run(self, record: RunRecord | None = None) -> RunRecord:
  if record is None:
@@ -63,10 +91,26 @@ def cancel_request_path(self, run_id: str) -> Path:
 
  async def persist_run(self, run_id: str) -> None:
  record = self._runs[run_id]
- run_dir = self.run_dir(run_id)
- lock = self._locks[run_id]
- with lock:
- (run_dir / "run.json").write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ path = self.run_dir(run_id) / "run.json"
+
+ # Check if disk is newer before overwriting
+ if path.exists():
+ try:
+ disk_mtime = path.stat().st_mtime
+ if self._mtimes.get(run_id, 0) < disk_mtime:
+ # Disk is newer, reload first to avoid overwriting clinical updates (like failed status)
+ disk_content = path.read_text(encoding="utf-8")
+ disk_record = RunRecord.model_validate_json(disk_content)
+ if disk_record.status in ("failed", "cancelled", "completed"):
+ # Don't overwrite terminal status
+ self._runs[run_id] = disk_record
+ self._mtimes[run_id] = disk_mtime
+ return
+ except Exception as e:
+ sync_logger.error(f"Error checking disk state for {run_id}: {e}")
+
+ path.write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ self._mtimes[run_id] = path.stat().st_mtime
 
  async def append_event(self, run_id: str, event: RunEvent) -> None:
  lock = self._locks[run_id]
@@ -112,12 +156,15 @@ def read_artifact_text(self, run_id: str, node_id: str, name: str) -> str:
  return self.artifact_path(run_id, node_id, name).read_text(encoding="utf-8")
 
  def get_run(self, run_id: str) -> RunRecord:
+ self._sync_runs()
  return self._runs[run_id]
 
  def list_runs(self) -> list[RunRecord]:
+ self._sync_runs()
  return sorted(self._runs.values(), key=lambda run: run.created_at, reverse=True)
 
  def get_events(self, run_id: str) -> list[RunEvent]:
+ self._sync_runs()
  return list(self._events_cache[run_id])
 
  async def subscribe(self, run_id: str) -> queue.Queue[RunEvent]:

agentflow/web/frontend/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

agentflow/web/frontend/README.md

@@ -0,0 +1,73 @@
+# React + TypeScript + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react) uses [Oxc](https://oxc.rs)
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react-swc) uses [SWC](https://swc.rs/)
+
+## React Compiler
+
+The React Compiler is not enabled on this template because of its impact on dev & build performances. To add it, see [this documentation](https://react.dev/learn/react-compiler/installation).
+
+## Expanding the ESLint configuration
+
+If you are developing a production application, we recommend updating the configuration to enable type-aware lint rules:
+
+```js
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+
+ // Remove tseslint.configs.recommended and replace with this
+ tseslint.configs.recommendedTypeChecked,
+ // Alternatively, use this for stricter rules
+ tseslint.configs.strictTypeChecked,
+ // Optionally, add this for stylistic rules
+ tseslint.configs.stylisticTypeChecked,
+
+ // Other configs...
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```
+
+You can also install [eslint-plugin-react-x](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-x) and [eslint-plugin-react-dom](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-dom) for React-specific lint rules:
+
+```js
+// eslint.config.js
+import reactX from 'eslint-plugin-react-x'
+import reactDom from 'eslint-plugin-react-dom'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+ // Enable lint rules for React
+ reactX.configs['recommended-typescript'],
+ // Enable lint rules for React DOM
+ reactDom.configs.recommended,
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```

agentflow/web/frontend/eslint.config.js

@@ -0,0 +1,23 @@
+import js from '@eslint/js'
+import globals from 'globals'
+import reactHooks from 'eslint-plugin-react-hooks'
+import reactRefresh from 'eslint-plugin-react-refresh'
+import tseslint from 'typescript-eslint'
+import { defineConfig, globalIgnores } from 'eslint/config'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ js.configs.recommended,
+ tseslint.configs.recommended,
+ reactHooks.configs.flat.recommended,
+ reactRefresh.configs.vite,
+ ],
+ languageOptions: {
+ ecmaVersion: 2020,
+ globals: globals.browser,
+ },
+ },
+])

agentflow/web/frontend/index.html

@@ -0,0 +1,13 @@
+<!doctype html>
+<htmllang="en">
+<head>
+<metacharset="UTF-8" />
+<linkrel="icon" type="image/svg+xml" href="/favicon.svg" />
+<metaname="viewport" content="width=device-width, initial-scale=1.0" />
+<title>frontend</title>
+</head>
+<body>
+<divid="root"></div>
+<scripttype="module" src="/src/main.tsx"></script>
+</body>
+</html>

.gitignore

@@ -10,4 +10,5 @@ test-results/
 .e2e-flaky
 dist/
 *.egg-info/
-.worktrees/
+.worktrees/
+uv.lock

Makefile

@@ -1,17 +1,26 @@
 .DEFAULT_GOAL := help
 
-.PHONY: help test smoke
+.PHONY: help test smoke install-ui build-ui
 
 PYTHON := $(if $(wildcard .venv/bin/python),.venv/bin/python,python3)
+FRONTEND_DIR := agentflow/web/frontend
 
 help:
 	@printf '%s\n' \
  'Available targets:' \
- ' test Run the test suite' \
- ' smoke Run the default smoke pipeline'
+ ' test Run the test suite' \
+ ' smoke Run the default smoke pipeline' \
+ ' install-ui Install frontend dependencies' \
+ ' build-ui Build the frontend dashboard'
 
 test:
 	$(PYTHON) -m pytest -q
 
 smoke:
 	$(PYTHON) -m agentflow run examples/airflow_like.py --output summary
+
+install-ui:
+	cd $(FRONTEND_DIR) && npm install
+
+build-ui:
+	cd $(FRONTEND_DIR) && npm run build

README.md

@@ -22,6 +22,12 @@ python3 -m venv .venv && . .venv/bin/activate
 pip install -e .[dev]
 ```
 
+To build the dashboard (optional, requires Node.js):
+```bash
+make install-ui
+make build-ui
+```
+
 ## Quick Start
 
 ```python

agentflow/app.py

@@ -84,11 +84,28 @@ def create_app(*, store: RunStore | None = None, orchestrator: Orchestrator | No
  app.state.orchestrator = orchestrator
 
  base_dir = os.path.join(os.path.dirname(__file__), "web")
+ frontend_dist = os.path.join(base_dir, "frontend", "dist")
+
+ # Mount Vite static assets
+ if os.path.isdir(frontend_dist):
+ assets_dir = os.path.join(frontend_dist, "assets")
+ if os.path.isdir(assets_dir):
+ app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+ # Public folder/static files inside dist
+ app.mount("/public", StaticFiles(directory=frontend_dist), name="public-dist")
+
+ # Legacy mounts and templates
  templates = Jinja2Templates(directory=os.path.join(base_dir, "templates"))
  app.mount("/static", StaticFiles(directory=os.path.join(base_dir, "static")), name="static")
 
  @app.get("/", response_class=HTMLResponse)
  async def index(request: Request) -> HTMLResponse:
+ react_index = os.path.join(frontend_dist, "index.html")
+ if os.path.isfile(react_index):
+ with open(react_index, "r") as f:
+ return HTMLResponse(content=f.read(), status_code=200)
+
  return templates.TemplateResponse(
  "index.html",
  {"request": request, "example": _load_default_web_example(), "base_dir": os.getcwd()},
@@ -153,6 +170,17 @@ async def get_artifact(run_id: str, node_id: str, name: str) -> PlainTextRespons
  raise HTTPException(status_code=404, detail="artifact not found") from exc
  return PlainTextResponse(content)
 
+ @app.get("/api/runs/{run_id}/scratchboard")
+ async def get_scratchboard(run_id: str) -> PlainTextResponse:
+ from agentflow.scratchboard import SCRATCHBOARD_FILENAME
+ try:
+ path = app.state.store.run_dir(run_id) / SCRATCHBOARD_FILENAME
+ if not path.exists():
+ return PlainTextResponse("")
+ return PlainTextResponse(path.read_text(encoding="utf-8"))
+ except Exception as exc:
+ raise HTTPException(status_code=404, detail="scratchboard not found") from exc
+
  @app.get("/api/runs/{run_id}/stream")
  async def stream_run(run_id: str):
  if run_id not in {run.id for run in app.state.store.list_runs()}:

agentflow/store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import queue
 import threading
 from collections import defaultdict
@@ -12,6 +13,9 @@
 from agentflow.specs import RunEvent, RunRecord
 from agentflow.utils import ensure_dir
 
+# Set up a dedicated logger for sync issues
+sync_logger = logging.getLogger("agentflow.sync")
+
 
 class RunStore:
  def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
@@ -20,24 +24,48 @@ def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
  self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
  self._subscribers: defaultdict[str, set[queue.Queue[RunEvent]]] = defaultdict(set)
  self._events_cache: defaultdict[str, list[RunEvent]] = defaultdict(list)
- self._load_existing_runs()
-
- def _load_existing_runs(self) -> None:
- for run_file in sorted(self.base_dir.glob("*/run.json")):
- run_id = run_file.parent.name
- try:
- run = RunRecord.model_validate_json(run_file.read_text(encoding="utf-8"))
- self._runs[run_id] = run
- events_path = run_file.parent / "events.jsonl"
- if events_path.exists():
- events = [
- RunEvent.model_validate_json(line)
- for line in events_path.read_text(encoding="utf-8").splitlines()
- if line.strip()
- ]
- self._events_cache[run_id] = events
- except (OSError, ValidationError, json.JSONDecodeError, KeyError):
- continue
+ self._mtimes: dict[str, float] = {}
+ self._sync_runs()
+
+ def _sync_runs(self) -> None:
+ """Synchronize in-memory cache with the filesystem."""
+ try:
+ run_files = list(self.base_dir.glob("*/run.json"))
+ for run_file in sorted(run_files):
+ run_id = run_file.parent.name
+ try:
+ mtime = run_file.stat().st_mtime
+ if run_id not in self._runs or self._mtimes.get(run_id, 0) < mtime:
+ content = run_file.read_text(encoding="utf-8")
+ if not content.strip():
+ continue
+
+ run = RunRecord.model_validate_json(content)
+ # Only update if status changed or it's new
+ if run_id not in self._runs or self._runs[run_id].status != run.status:
+ sync_logger.debug(f"Syncing run {run_id}: status {run.status}")
+
+ self._runs[run_id] = run
+ self._mtimes[run_id] = mtime
+
+ # Sync events
+ events_path = run_file.parent / "events.jsonl"
+ if events_path.exists():
+ event_mtime = events_path.stat().st_mtime
+ if self._mtimes.get(f"{run_id}_events", 0) < event_mtime:
+ events = [
+ RunEvent.model_validate_json(line)
+ for line in events_path.read_text(encoding="utf-8").splitlines()
+ if line.strip()
+ ]
+ self._events_cache[run_id] = events
+ self._mtimes[f"{run_id}_events"] = event_mtime
+ except (OSError, ValidationError, json.JSONDecodeError, KeyError) as e:
+ sync_logger.error(f"Failed to sync {run_id}: {e}")
+ continue
+ except OSError as e:
+ sync_logger.error(f"Sync error: {e}")
+ pass
 
  async def create_run(self, record: RunRecord | None = None) -> RunRecord:
  if record is None:
@@ -63,10 +91,26 @@ def cancel_request_path(self, run_id: str) -> Path:
 
  async def persist_run(self, run_id: str) -> None:
  record = self._runs[run_id]
- run_dir = self.run_dir(run_id)
- lock = self._locks[run_id]
- with lock:
- (run_dir / "run.json").write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ path = self.run_dir(run_id) / "run.json"
+
+ # Check if disk is newer before overwriting
+ if path.exists():
+ try:
+ disk_mtime = path.stat().st_mtime
+ if self._mtimes.get(run_id, 0) < disk_mtime:
+ # Disk is newer, reload first to avoid overwriting clinical updates (like failed status)
+ disk_content = path.read_text(encoding="utf-8")
+ disk_record = RunRecord.model_validate_json(disk_content)
+ if disk_record.status in ("failed", "cancelled", "completed"):
+ # Don't overwrite terminal status
+ self._runs[run_id] = disk_record
+ self._mtimes[run_id] = disk_mtime
+ return
+ except Exception as e:
+ sync_logger.error(f"Error checking disk state for {run_id}: {e}")
+
+ path.write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ self._mtimes[run_id] = path.stat().st_mtime
 
  async def append_event(self, run_id: str, event: RunEvent) -> None:
  lock = self._locks[run_id]
@@ -112,12 +156,15 @@ def read_artifact_text(self, run_id: str, node_id: str, name: str) -> str:
  return self.artifact_path(run_id, node_id, name).read_text(encoding="utf-8")
 
  def get_run(self, run_id: str) -> RunRecord:
+ self._sync_runs()
  return self._runs[run_id]
 
  def list_runs(self) -> list[RunRecord]:
+ self._sync_runs()
  return sorted(self._runs.values(), key=lambda run: run.created_at, reverse=True)
 
  def get_events(self, run_id: str) -> list[RunEvent]:
+ self._sync_runs()
  return list(self._events_cache[run_id])
 
  async def subscribe(self, run_id: str) -> queue.Queue[RunEvent]:

agentflow/web/frontend/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

agentflow/web/frontend/README.md

@@ -0,0 +1,73 @@
+# React + TypeScript + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react) uses [Oxc](https://oxc.rs)
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react-swc) uses [SWC](https://swc.rs/)
+
+## React Compiler
+
+The React Compiler is not enabled on this template because of its impact on dev & build performances. To add it, see [this documentation](https://react.dev/learn/react-compiler/installation).
+
+## Expanding the ESLint configuration
+
+If you are developing a production application, we recommend updating the configuration to enable type-aware lint rules:
+
+```js
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+
+ // Remove tseslint.configs.recommended and replace with this
+ tseslint.configs.recommendedTypeChecked,
+ // Alternatively, use this for stricter rules
+ tseslint.configs.strictTypeChecked,
+ // Optionally, add this for stylistic rules
+ tseslint.configs.stylisticTypeChecked,
+
+ // Other configs...
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```
+
+You can also install [eslint-plugin-react-x](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-x) and [eslint-plugin-react-dom](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-dom) for React-specific lint rules:
+
+```js
+// eslint.config.js
+import reactX from 'eslint-plugin-react-x'
+import reactDom from 'eslint-plugin-react-dom'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+ // Enable lint rules for React
+ reactX.configs['recommended-typescript'],
+ // Enable lint rules for React DOM
+ reactDom.configs.recommended,
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```

agentflow/web/frontend/eslint.config.js

@@ -0,0 +1,23 @@
+import js from '@eslint/js'
+import globals from 'globals'
+import reactHooks from 'eslint-plugin-react-hooks'
+import reactRefresh from 'eslint-plugin-react-refresh'
+import tseslint from 'typescript-eslint'
+import { defineConfig, globalIgnores } from 'eslint/config'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ js.configs.recommended,
+ tseslint.configs.recommended,
+ reactHooks.configs.flat.recommended,
+ reactRefresh.configs.vite,
+ ],
+ languageOptions: {
+ ecmaVersion: 2020,
+ globals: globals.browser,
+ },
+ },
+])

agentflow/web/frontend/index.html

@@ -0,0 +1,13 @@
+<!doctype html>
+<htmllang="en">
+<head>
+<metacharset="UTF-8" />
+<linkrel="icon" type="image/svg+xml" href="/favicon.svg" />
+<metaname="viewport" content="width=device-width, initial-scale=1.0" />
+<title>frontend</title>
+</head>
+<body>
+<divid="root"></div>
+<scripttype="module" src="/src/main.tsx"></script>
+</body>
+</html>

.gitignore

@@ -10,4 +10,5 @@ test-results/
 .e2e-flaky
 dist/
 *.egg-info/
-.worktrees/
+.worktrees/
+uv.lock

Makefile

@@ -1,17 +1,26 @@
 .DEFAULT_GOAL := help
 
-.PHONY: help test smoke
+.PHONY: help test smoke install-ui build-ui
 
 PYTHON := $(if $(wildcard .venv/bin/python),.venv/bin/python,python3)
+FRONTEND_DIR := agentflow/web/frontend
 
 help:
 	@printf '%s\n' \
  'Available targets:' \
- ' test Run the test suite' \
- ' smoke Run the default smoke pipeline'
+ ' test Run the test suite' \
+ ' smoke Run the default smoke pipeline' \
+ ' install-ui Install frontend dependencies' \
+ ' build-ui Build the frontend dashboard'
 
 test:
 	$(PYTHON) -m pytest -q
 
 smoke:
 	$(PYTHON) -m agentflow run examples/airflow_like.py --output summary
+
+install-ui:
+	cd $(FRONTEND_DIR) && npm install
+
+build-ui:
+	cd $(FRONTEND_DIR) && npm run build

README.md

@@ -22,6 +22,12 @@ python3 -m venv .venv && . .venv/bin/activate
 pip install -e .[dev]
 ```
 
+To build the dashboard (optional, requires Node.js):
+```bash
+make install-ui
+make build-ui
+```
+
 ## Quick Start
 
 ```python

agentflow/app.py

@@ -84,11 +84,28 @@ def create_app(*, store: RunStore | None = None, orchestrator: Orchestrator | No
  app.state.orchestrator = orchestrator
 
  base_dir = os.path.join(os.path.dirname(__file__), "web")
+ frontend_dist = os.path.join(base_dir, "frontend", "dist")
+
+ # Mount Vite static assets
+ if os.path.isdir(frontend_dist):
+ assets_dir = os.path.join(frontend_dist, "assets")
+ if os.path.isdir(assets_dir):
+ app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+ # Public folder/static files inside dist
+ app.mount("/public", StaticFiles(directory=frontend_dist), name="public-dist")
+
+ # Legacy mounts and templates
  templates = Jinja2Templates(directory=os.path.join(base_dir, "templates"))
  app.mount("/static", StaticFiles(directory=os.path.join(base_dir, "static")), name="static")
 
  @app.get("/", response_class=HTMLResponse)
  async def index(request: Request) -> HTMLResponse:
+ react_index = os.path.join(frontend_dist, "index.html")
+ if os.path.isfile(react_index):
+ with open(react_index, "r") as f:
+ return HTMLResponse(content=f.read(), status_code=200)
+
  return templates.TemplateResponse(
  "index.html",
  {"request": request, "example": _load_default_web_example(), "base_dir": os.getcwd()},
@@ -153,6 +170,17 @@ async def get_artifact(run_id: str, node_id: str, name: str) -> PlainTextRespons
  raise HTTPException(status_code=404, detail="artifact not found") from exc
  return PlainTextResponse(content)
 
+ @app.get("/api/runs/{run_id}/scratchboard")
+ async def get_scratchboard(run_id: str) -> PlainTextResponse:
+ from agentflow.scratchboard import SCRATCHBOARD_FILENAME
+ try:
+ path = app.state.store.run_dir(run_id) / SCRATCHBOARD_FILENAME
+ if not path.exists():
+ return PlainTextResponse("")
+ return PlainTextResponse(path.read_text(encoding="utf-8"))
+ except Exception as exc:
+ raise HTTPException(status_code=404, detail="scratchboard not found") from exc
+
  @app.get("/api/runs/{run_id}/stream")
  async def stream_run(run_id: str):
  if run_id not in {run.id for run in app.state.store.list_runs()}:

agentflow/store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import queue
 import threading
 from collections import defaultdict
@@ -12,6 +13,9 @@
 from agentflow.specs import RunEvent, RunRecord
 from agentflow.utils import ensure_dir
 
+# Set up a dedicated logger for sync issues
+sync_logger = logging.getLogger("agentflow.sync")
+
 
 class RunStore:
  def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
@@ -20,24 +24,48 @@ def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
  self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
  self._subscribers: defaultdict[str, set[queue.Queue[RunEvent]]] = defaultdict(set)
  self._events_cache: defaultdict[str, list[RunEvent]] = defaultdict(list)
- self._load_existing_runs()
-
- def _load_existing_runs(self) -> None:
- for run_file in sorted(self.base_dir.glob("*/run.json")):
- run_id = run_file.parent.name
- try:
- run = RunRecord.model_validate_json(run_file.read_text(encoding="utf-8"))
- self._runs[run_id] = run
- events_path = run_file.parent / "events.jsonl"
- if events_path.exists():
- events = [
- RunEvent.model_validate_json(line)
- for line in events_path.read_text(encoding="utf-8").splitlines()
- if line.strip()
- ]
- self._events_cache[run_id] = events
- except (OSError, ValidationError, json.JSONDecodeError, KeyError):
- continue
+ self._mtimes: dict[str, float] = {}
+ self._sync_runs()
+
+ def _sync_runs(self) -> None:
+ """Synchronize in-memory cache with the filesystem."""
+ try:
+ run_files = list(self.base_dir.glob("*/run.json"))
+ for run_file in sorted(run_files):
+ run_id = run_file.parent.name
+ try:
+ mtime = run_file.stat().st_mtime
+ if run_id not in self._runs or self._mtimes.get(run_id, 0) < mtime:
+ content = run_file.read_text(encoding="utf-8")
+ if not content.strip():
+ continue
+
+ run = RunRecord.model_validate_json(content)
+ # Only update if status changed or it's new
+ if run_id not in self._runs or self._runs[run_id].status != run.status:
+ sync_logger.debug(f"Syncing run {run_id}: status {run.status}")
+
+ self._runs[run_id] = run
+ self._mtimes[run_id] = mtime
+
+ # Sync events
+ events_path = run_file.parent / "events.jsonl"
+ if events_path.exists():
+ event_mtime = events_path.stat().st_mtime
+ if self._mtimes.get(f"{run_id}_events", 0) < event_mtime:
+ events = [
+ RunEvent.model_validate_json(line)
+ for line in events_path.read_text(encoding="utf-8").splitlines()
+ if line.strip()
+ ]
+ self._events_cache[run_id] = events
+ self._mtimes[f"{run_id}_events"] = event_mtime
+ except (OSError, ValidationError, json.JSONDecodeError, KeyError) as e:
+ sync_logger.error(f"Failed to sync {run_id}: {e}")
+ continue
+ except OSError as e:
+ sync_logger.error(f"Sync error: {e}")
+ pass
 
  async def create_run(self, record: RunRecord | None = None) -> RunRecord:
  if record is None:
@@ -63,10 +91,26 @@ def cancel_request_path(self, run_id: str) -> Path:
 
  async def persist_run(self, run_id: str) -> None:
  record = self._runs[run_id]
- run_dir = self.run_dir(run_id)
- lock = self._locks[run_id]
- with lock:
- (run_dir / "run.json").write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ path = self.run_dir(run_id) / "run.json"
+
+ # Check if disk is newer before overwriting
+ if path.exists():
+ try:
+ disk_mtime = path.stat().st_mtime
+ if self._mtimes.get(run_id, 0) < disk_mtime:
+ # Disk is newer, reload first to avoid overwriting clinical updates (like failed status)
+ disk_content = path.read_text(encoding="utf-8")
+ disk_record = RunRecord.model_validate_json(disk_content)
+ if disk_record.status in ("failed", "cancelled", "completed"):
+ # Don't overwrite terminal status
+ self._runs[run_id] = disk_record
+ self._mtimes[run_id] = disk_mtime
+ return
+ except Exception as e:
+ sync_logger.error(f"Error checking disk state for {run_id}: {e}")
+
+ path.write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ self._mtimes[run_id] = path.stat().st_mtime
 
  async def append_event(self, run_id: str, event: RunEvent) -> None:
  lock = self._locks[run_id]
@@ -112,12 +156,15 @@ def read_artifact_text(self, run_id: str, node_id: str, name: str) -> str:
  return self.artifact_path(run_id, node_id, name).read_text(encoding="utf-8")
 
  def get_run(self, run_id: str) -> RunRecord:
+ self._sync_runs()
  return self._runs[run_id]
 
  def list_runs(self) -> list[RunRecord]:
+ self._sync_runs()
  return sorted(self._runs.values(), key=lambda run: run.created_at, reverse=True)
 
  def get_events(self, run_id: str) -> list[RunEvent]:
+ self._sync_runs()
  return list(self._events_cache[run_id])
 
  async def subscribe(self, run_id: str) -> queue.Queue[RunEvent]:

agentflow/web/frontend/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

agentflow/web/frontend/README.md

@@ -0,0 +1,73 @@
+# React + TypeScript + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react) uses [Oxc](https://oxc.rs)
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react-swc) uses [SWC](https://swc.rs/)
+
+## React Compiler
+
+The React Compiler is not enabled on this template because of its impact on dev & build performances. To add it, see [this documentation](https://react.dev/learn/react-compiler/installation).
+
+## Expanding the ESLint configuration
+
+If you are developing a production application, we recommend updating the configuration to enable type-aware lint rules:
+
+```js
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+
+ // Remove tseslint.configs.recommended and replace with this
+ tseslint.configs.recommendedTypeChecked,
+ // Alternatively, use this for stricter rules
+ tseslint.configs.strictTypeChecked,
+ // Optionally, add this for stylistic rules
+ tseslint.configs.stylisticTypeChecked,
+
+ // Other configs...
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```
+
+You can also install [eslint-plugin-react-x](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-x) and [eslint-plugin-react-dom](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-dom) for React-specific lint rules:
+
+```js
+// eslint.config.js
+import reactX from 'eslint-plugin-react-x'
+import reactDom from 'eslint-plugin-react-dom'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+ // Enable lint rules for React
+ reactX.configs['recommended-typescript'],
+ // Enable lint rules for React DOM
+ reactDom.configs.recommended,
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```

agentflow/web/frontend/eslint.config.js

@@ -0,0 +1,23 @@
+import js from '@eslint/js'
+import globals from 'globals'
+import reactHooks from 'eslint-plugin-react-hooks'
+import reactRefresh from 'eslint-plugin-react-refresh'
+import tseslint from 'typescript-eslint'
+import { defineConfig, globalIgnores } from 'eslint/config'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ js.configs.recommended,
+ tseslint.configs.recommended,
+ reactHooks.configs.flat.recommended,
+ reactRefresh.configs.vite,
+ ],
+ languageOptions: {
+ ecmaVersion: 2020,
+ globals: globals.browser,
+ },
+ },
+])

agentflow/web/frontend/index.html

@@ -0,0 +1,13 @@
+<!doctype html>
+<htmllang="en">
+<head>
+<metacharset="UTF-8" />
+<linkrel="icon" type="image/svg+xml" href="/favicon.svg" />
+<metaname="viewport" content="width=device-width, initial-scale=1.0" />
+<title>frontend</title>
+</head>
+<body>
+<divid="root"></div>
+<scripttype="module" src="/src/main.tsx"></script>
+</body>
+</html>

.gitignore

@@ -10,4 +10,5 @@ test-results/
 .e2e-flaky
 dist/
 *.egg-info/
-.worktrees/
+.worktrees/
+uv.lock

Makefile

@@ -1,17 +1,26 @@
 .DEFAULT_GOAL := help
 
-.PHONY: help test smoke
+.PHONY: help test smoke install-ui build-ui
 
 PYTHON := $(if $(wildcard .venv/bin/python),.venv/bin/python,python3)
+FRONTEND_DIR := agentflow/web/frontend
 
 help:
 	@printf '%s\n' \
  'Available targets:' \
- ' test Run the test suite' \
- ' smoke Run the default smoke pipeline'
+ ' test Run the test suite' \
+ ' smoke Run the default smoke pipeline' \
+ ' install-ui Install frontend dependencies' \
+ ' build-ui Build the frontend dashboard'
 
 test:
 	$(PYTHON) -m pytest -q
 
 smoke:
 	$(PYTHON) -m agentflow run examples/airflow_like.py --output summary
+
+install-ui:
+	cd $(FRONTEND_DIR) && npm install
+
+build-ui:
+	cd $(FRONTEND_DIR) && npm run build

README.md

@@ -22,6 +22,12 @@ python3 -m venv .venv && . .venv/bin/activate
 pip install -e .[dev]
 ```
 
+To build the dashboard (optional, requires Node.js):
+```bash
+make install-ui
+make build-ui
+```
+
 ## Quick Start
 
 ```python

agentflow/app.py

@@ -84,11 +84,28 @@ def create_app(*, store: RunStore | None = None, orchestrator: Orchestrator | No
  app.state.orchestrator = orchestrator
 
  base_dir = os.path.join(os.path.dirname(__file__), "web")
+ frontend_dist = os.path.join(base_dir, "frontend", "dist")
+
+ # Mount Vite static assets
+ if os.path.isdir(frontend_dist):
+ assets_dir = os.path.join(frontend_dist, "assets")
+ if os.path.isdir(assets_dir):
+ app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+ # Public folder/static files inside dist
+ app.mount("/public", StaticFiles(directory=frontend_dist), name="public-dist")
+
+ # Legacy mounts and templates
  templates = Jinja2Templates(directory=os.path.join(base_dir, "templates"))
  app.mount("/static", StaticFiles(directory=os.path.join(base_dir, "static")), name="static")
 
  @app.get("/", response_class=HTMLResponse)
  async def index(request: Request) -> HTMLResponse:
+ react_index = os.path.join(frontend_dist, "index.html")
+ if os.path.isfile(react_index):
+ with open(react_index, "r") as f:
+ return HTMLResponse(content=f.read(), status_code=200)
+
  return templates.TemplateResponse(
  "index.html",
  {"request": request, "example": _load_default_web_example(), "base_dir": os.getcwd()},
@@ -153,6 +170,17 @@ async def get_artifact(run_id: str, node_id: str, name: str) -> PlainTextRespons
  raise HTTPException(status_code=404, detail="artifact not found") from exc
  return PlainTextResponse(content)
 
+ @app.get("/api/runs/{run_id}/scratchboard")
+ async def get_scratchboard(run_id: str) -> PlainTextResponse:
+ from agentflow.scratchboard import SCRATCHBOARD_FILENAME
+ try:
+ path = app.state.store.run_dir(run_id) / SCRATCHBOARD_FILENAME
+ if not path.exists():
+ return PlainTextResponse("")
+ return PlainTextResponse(path.read_text(encoding="utf-8"))
+ except Exception as exc:
+ raise HTTPException(status_code=404, detail="scratchboard not found") from exc
+
  @app.get("/api/runs/{run_id}/stream")
  async def stream_run(run_id: str):
  if run_id not in {run.id for run in app.state.store.list_runs()}:

agentflow/store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import queue
 import threading
 from collections import defaultdict
@@ -12,6 +13,9 @@
 from agentflow.specs import RunEvent, RunRecord
 from agentflow.utils import ensure_dir
 
+# Set up a dedicated logger for sync issues
+sync_logger = logging.getLogger("agentflow.sync")
+
 
 class RunStore:
  def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
@@ -20,24 +24,48 @@ def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
  self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
  self._subscribers: defaultdict[str, set[queue.Queue[RunEvent]]] = defaultdict(set)
  self._events_cache: defaultdict[str, list[RunEvent]] = defaultdict(list)
- self._load_existing_runs()
-
- def _load_existing_runs(self) -> None:
- for run_file in sorted(self.base_dir.glob("*/run.json")):
- run_id = run_file.parent.name
- try:
- run = RunRecord.model_validate_json(run_file.read_text(encoding="utf-8"))
- self._runs[run_id] = run
- events_path = run_file.parent / "events.jsonl"
- if events_path.exists():
- events = [
- RunEvent.model_validate_json(line)
- for line in events_path.read_text(encoding="utf-8").splitlines()
- if line.strip()
- ]
- self._events_cache[run_id] = events
- except (OSError, ValidationError, json.JSONDecodeError, KeyError):
- continue
+ self._mtimes: dict[str, float] = {}
+ self._sync_runs()
+
+ def _sync_runs(self) -> None:
+ """Synchronize in-memory cache with the filesystem."""
+ try:
+ run_files = list(self.base_dir.glob("*/run.json"))
+ for run_file in sorted(run_files):
+ run_id = run_file.parent.name
+ try:
+ mtime = run_file.stat().st_mtime
+ if run_id not in self._runs or self._mtimes.get(run_id, 0) < mtime:
+ content = run_file.read_text(encoding="utf-8")
+ if not content.strip():
+ continue
+
+ run = RunRecord.model_validate_json(content)
+ # Only update if status changed or it's new
+ if run_id not in self._runs or self._runs[run_id].status != run.status:
+ sync_logger.debug(f"Syncing run {run_id}: status {run.status}")
+
+ self._runs[run_id] = run
+ self._mtimes[run_id] = mtime
+
+ # Sync events
+ events_path = run_file.parent / "events.jsonl"
+ if events_path.exists():
+ event_mtime = events_path.stat().st_mtime
+ if self._mtimes.get(f"{run_id}_events", 0) < event_mtime:
+ events = [
+ RunEvent.model_validate_json(line)
+ for line in events_path.read_text(encoding="utf-8").splitlines()
+ if line.strip()
+ ]
+ self._events_cache[run_id] = events
+ self._mtimes[f"{run_id}_events"] = event_mtime
+ except (OSError, ValidationError, json.JSONDecodeError, KeyError) as e:
+ sync_logger.error(f"Failed to sync {run_id}: {e}")
+ continue
+ except OSError as e:
+ sync_logger.error(f"Sync error: {e}")
+ pass
 
  async def create_run(self, record: RunRecord | None = None) -> RunRecord:
  if record is None:
@@ -63,10 +91,26 @@ def cancel_request_path(self, run_id: str) -> Path:
 
  async def persist_run(self, run_id: str) -> None:
  record = self._runs[run_id]
- run_dir = self.run_dir(run_id)
- lock = self._locks[run_id]
- with lock:
- (run_dir / "run.json").write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ path = self.run_dir(run_id) / "run.json"
+
+ # Check if disk is newer before overwriting
+ if path.exists():
+ try:
+ disk_mtime = path.stat().st_mtime
+ if self._mtimes.get(run_id, 0) < disk_mtime:
+ # Disk is newer, reload first to avoid overwriting clinical updates (like failed status)
+ disk_content = path.read_text(encoding="utf-8")
+ disk_record = RunRecord.model_validate_json(disk_content)
+ if disk_record.status in ("failed", "cancelled", "completed"):
+ # Don't overwrite terminal status
+ self._runs[run_id] = disk_record
+ self._mtimes[run_id] = disk_mtime
+ return
+ except Exception as e:
+ sync_logger.error(f"Error checking disk state for {run_id}: {e}")
+
+ path.write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ self._mtimes[run_id] = path.stat().st_mtime
 
  async def append_event(self, run_id: str, event: RunEvent) -> None:
  lock = self._locks[run_id]
@@ -112,12 +156,15 @@ def read_artifact_text(self, run_id: str, node_id: str, name: str) -> str:
  return self.artifact_path(run_id, node_id, name).read_text(encoding="utf-8")
 
  def get_run(self, run_id: str) -> RunRecord:
+ self._sync_runs()
  return self._runs[run_id]
 
  def list_runs(self) -> list[RunRecord]:
+ self._sync_runs()
  return sorted(self._runs.values(), key=lambda run: run.created_at, reverse=True)
 
  def get_events(self, run_id: str) -> list[RunEvent]:
+ self._sync_runs()
  return list(self._events_cache[run_id])
 
  async def subscribe(self, run_id: str) -> queue.Queue[RunEvent]:

agentflow/web/frontend/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

agentflow/web/frontend/README.md

@@ -0,0 +1,73 @@
+# React + TypeScript + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react) uses [Oxc](https://oxc.rs)
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react-swc) uses [SWC](https://swc.rs/)
+
+## React Compiler
+
+The React Compiler is not enabled on this template because of its impact on dev & build performances. To add it, see [this documentation](https://react.dev/learn/react-compiler/installation).
+
+## Expanding the ESLint configuration
+
+If you are developing a production application, we recommend updating the configuration to enable type-aware lint rules:
+
+```js
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+
+ // Remove tseslint.configs.recommended and replace with this
+ tseslint.configs.recommendedTypeChecked,
+ // Alternatively, use this for stricter rules
+ tseslint.configs.strictTypeChecked,
+ // Optionally, add this for stylistic rules
+ tseslint.configs.stylisticTypeChecked,
+
+ // Other configs...
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```
+
+You can also install [eslint-plugin-react-x](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-x) and [eslint-plugin-react-dom](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-dom) for React-specific lint rules:
+
+```js
+// eslint.config.js
+import reactX from 'eslint-plugin-react-x'
+import reactDom from 'eslint-plugin-react-dom'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+ // Enable lint rules for React
+ reactX.configs['recommended-typescript'],
+ // Enable lint rules for React DOM
+ reactDom.configs.recommended,
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```

agentflow/web/frontend/eslint.config.js

@@ -0,0 +1,23 @@
+import js from '@eslint/js'
+import globals from 'globals'
+import reactHooks from 'eslint-plugin-react-hooks'
+import reactRefresh from 'eslint-plugin-react-refresh'
+import tseslint from 'typescript-eslint'
+import { defineConfig, globalIgnores } from 'eslint/config'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ js.configs.recommended,
+ tseslint.configs.recommended,
+ reactHooks.configs.flat.recommended,
+ reactRefresh.configs.vite,
+ ],
+ languageOptions: {
+ ecmaVersion: 2020,
+ globals: globals.browser,
+ },
+ },
+])

agentflow/web/frontend/index.html

@@ -0,0 +1,13 @@
+<!doctype html>
+<htmllang="en">
+<head>
+<metacharset="UTF-8" />
+<linkrel="icon" type="image/svg+xml" href="/favicon.svg" />
+<metaname="viewport" content="width=device-width, initial-scale=1.0" />
+<title>frontend</title>
+</head>
+<body>
+<divid="root"></div>
+<scripttype="module" src="/src/main.tsx"></script>
+</body>
+</html>

.gitignore

@@ -10,4 +10,5 @@ test-results/
 .e2e-flaky
 dist/
 *.egg-info/
-.worktrees/
+.worktrees/
+uv.lock

Makefile

@@ -1,17 +1,26 @@
 .DEFAULT_GOAL := help
 
-.PHONY: help test smoke
+.PHONY: help test smoke install-ui build-ui
 
 PYTHON := $(if $(wildcard .venv/bin/python),.venv/bin/python,python3)
+FRONTEND_DIR := agentflow/web/frontend
 
 help:
 	@printf '%s\n' \
  'Available targets:' \
- ' test Run the test suite' \
- ' smoke Run the default smoke pipeline'
+ ' test Run the test suite' \
+ ' smoke Run the default smoke pipeline' \
+ ' install-ui Install frontend dependencies' \
+ ' build-ui Build the frontend dashboard'
 
 test:
 	$(PYTHON) -m pytest -q
 
 smoke:
 	$(PYTHON) -m agentflow run examples/airflow_like.py --output summary
+
+install-ui:
+	cd $(FRONTEND_DIR) && npm install
+
+build-ui:
+	cd $(FRONTEND_DIR) && npm run build

README.md

@@ -22,6 +22,12 @@ python3 -m venv .venv && . .venv/bin/activate
 pip install -e .[dev]
 ```
 
+To build the dashboard (optional, requires Node.js):
+```bash
+make install-ui
+make build-ui
+```
+
 ## Quick Start
 
 ```python

agentflow/app.py

@@ -84,11 +84,28 @@ def create_app(*, store: RunStore | None = None, orchestrator: Orchestrator | No
  app.state.orchestrator = orchestrator
 
  base_dir = os.path.join(os.path.dirname(__file__), "web")
+ frontend_dist = os.path.join(base_dir, "frontend", "dist")
+
+ # Mount Vite static assets
+ if os.path.isdir(frontend_dist):
+ assets_dir = os.path.join(frontend_dist, "assets")
+ if os.path.isdir(assets_dir):
+ app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+ # Public folder/static files inside dist
+ app.mount("/public", StaticFiles(directory=frontend_dist), name="public-dist")
+
+ # Legacy mounts and templates
  templates = Jinja2Templates(directory=os.path.join(base_dir, "templates"))
  app.mount("/static", StaticFiles(directory=os.path.join(base_dir, "static")), name="static")
 
  @app.get("/", response_class=HTMLResponse)
  async def index(request: Request) -> HTMLResponse:
+ react_index = os.path.join(frontend_dist, "index.html")
+ if os.path.isfile(react_index):
+ with open(react_index, "r") as f:
+ return HTMLResponse(content=f.read(), status_code=200)
+
  return templates.TemplateResponse(
  "index.html",
  {"request": request, "example": _load_default_web_example(), "base_dir": os.getcwd()},
@@ -153,6 +170,17 @@ async def get_artifact(run_id: str, node_id: str, name: str) -> PlainTextRespons
  raise HTTPException(status_code=404, detail="artifact not found") from exc
  return PlainTextResponse(content)
 
+ @app.get("/api/runs/{run_id}/scratchboard")
+ async def get_scratchboard(run_id: str) -> PlainTextResponse:
+ from agentflow.scratchboard import SCRATCHBOARD_FILENAME
+ try:
+ path = app.state.store.run_dir(run_id) / SCRATCHBOARD_FILENAME
+ if not path.exists():
+ return PlainTextResponse("")
+ return PlainTextResponse(path.read_text(encoding="utf-8"))
+ except Exception as exc:
+ raise HTTPException(status_code=404, detail="scratchboard not found") from exc
+
  @app.get("/api/runs/{run_id}/stream")
  async def stream_run(run_id: str):
  if run_id not in {run.id for run in app.state.store.list_runs()}:

agentflow/store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import queue
 import threading
 from collections import defaultdict
@@ -12,6 +13,9 @@
 from agentflow.specs import RunEvent, RunRecord
 from agentflow.utils import ensure_dir
 
+# Set up a dedicated logger for sync issues
+sync_logger = logging.getLogger("agentflow.sync")
+
 
 class RunStore:
  def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
@@ -20,24 +24,48 @@ def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
  self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
  self._subscribers: defaultdict[str, set[queue.Queue[RunEvent]]] = defaultdict(set)
  self._events_cache: defaultdict[str, list[RunEvent]] = defaultdict(list)
- self._load_existing_runs()
-
- def _load_existing_runs(self) -> None:
- for run_file in sorted(self.base_dir.glob("*/run.json")):
- run_id = run_file.parent.name
- try:
- run = RunRecord.model_validate_json(run_file.read_text(encoding="utf-8"))
- self._runs[run_id] = run
- events_path = run_file.parent / "events.jsonl"
- if events_path.exists():
- events = [
- RunEvent.model_validate_json(line)
- for line in events_path.read_text(encoding="utf-8").splitlines()
- if line.strip()
- ]
- self._events_cache[run_id] = events
- except (OSError, ValidationError, json.JSONDecodeError, KeyError):
- continue
+ self._mtimes: dict[str, float] = {}
+ self._sync_runs()
+
+ def _sync_runs(self) -> None:
+ """Synchronize in-memory cache with the filesystem."""
+ try:
+ run_files = list(self.base_dir.glob("*/run.json"))
+ for run_file in sorted(run_files):
+ run_id = run_file.parent.name
+ try:
+ mtime = run_file.stat().st_mtime
+ if run_id not in self._runs or self._mtimes.get(run_id, 0) < mtime:
+ content = run_file.read_text(encoding="utf-8")
+ if not content.strip():
+ continue
+
+ run = RunRecord.model_validate_json(content)
+ # Only update if status changed or it's new
+ if run_id not in self._runs or self._runs[run_id].status != run.status:
+ sync_logger.debug(f"Syncing run {run_id}: status {run.status}")
+
+ self._runs[run_id] = run
+ self._mtimes[run_id] = mtime
+
+ # Sync events
+ events_path = run_file.parent / "events.jsonl"
+ if events_path.exists():
+ event_mtime = events_path.stat().st_mtime
+ if self._mtimes.get(f"{run_id}_events", 0) < event_mtime:
+ events = [
+ RunEvent.model_validate_json(line)
+ for line in events_path.read_text(encoding="utf-8").splitlines()
+ if line.strip()
+ ]
+ self._events_cache[run_id] = events
+ self._mtimes[f"{run_id}_events"] = event_mtime
+ except (OSError, ValidationError, json.JSONDecodeError, KeyError) as e:
+ sync_logger.error(f"Failed to sync {run_id}: {e}")
+ continue
+ except OSError as e:
+ sync_logger.error(f"Sync error: {e}")
+ pass
 
  async def create_run(self, record: RunRecord | None = None) -> RunRecord:
  if record is None:
@@ -63,10 +91,26 @@ def cancel_request_path(self, run_id: str) -> Path:
 
  async def persist_run(self, run_id: str) -> None:
  record = self._runs[run_id]
- run_dir = self.run_dir(run_id)
- lock = self._locks[run_id]
- with lock:
- (run_dir / "run.json").write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ path = self.run_dir(run_id) / "run.json"
+
+ # Check if disk is newer before overwriting
+ if path.exists():
+ try:
+ disk_mtime = path.stat().st_mtime
+ if self._mtimes.get(run_id, 0) < disk_mtime:
+ # Disk is newer, reload first to avoid overwriting clinical updates (like failed status)
+ disk_content = path.read_text(encoding="utf-8")
+ disk_record = RunRecord.model_validate_json(disk_content)
+ if disk_record.status in ("failed", "cancelled", "completed"):
+ # Don't overwrite terminal status
+ self._runs[run_id] = disk_record
+ self._mtimes[run_id] = disk_mtime
+ return
+ except Exception as e:
+ sync_logger.error(f"Error checking disk state for {run_id}: {e}")
+
+ path.write_text(record.model_dump_json(indent=2), encoding="utf-8")
+ self._mtimes[run_id] = path.stat().st_mtime
 
  async def append_event(self, run_id: str, event: RunEvent) -> None:
  lock = self._locks[run_id]
@@ -112,12 +156,15 @@ def read_artifact_text(self, run_id: str, node_id: str, name: str) -> str:
  return self.artifact_path(run_id, node_id, name).read_text(encoding="utf-8")
 
  def get_run(self, run_id: str) -> RunRecord:
+ self._sync_runs()
  return self._runs[run_id]
 
  def list_runs(self) -> list[RunRecord]:
+ self._sync_runs()
  return sorted(self._runs.values(), key=lambda run: run.created_at, reverse=True)
 
  def get_events(self, run_id: str) -> list[RunEvent]:
+ self._sync_runs()
  return list(self._events_cache[run_id])
 
  async def subscribe(self, run_id: str) -> queue.Queue[RunEvent]:

agentflow/web/frontend/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

agentflow/web/frontend/README.md

@@ -0,0 +1,73 @@
+# React + TypeScript + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react) uses [Oxc](https://oxc.rs)
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react-swc) uses [SWC](https://swc.rs/)
+
+## React Compiler
+
+The React Compiler is not enabled on this template because of its impact on dev & build performances. To add it, see [this documentation](https://react.dev/learn/react-compiler/installation).
+
+## Expanding the ESLint configuration
+
+If you are developing a production application, we recommend updating the configuration to enable type-aware lint rules:
+
+```js
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+
+ // Remove tseslint.configs.recommended and replace with this
+ tseslint.configs.recommendedTypeChecked,
+ // Alternatively, use this for stricter rules
+ tseslint.configs.strictTypeChecked,
+ // Optionally, add this for stylistic rules
+ tseslint.configs.stylisticTypeChecked,
+
+ // Other configs...
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```
+
+You can also install [eslint-plugin-react-x](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-x) and [eslint-plugin-react-dom](https://github.com/Rel1cx/eslint-react/tree/main/packages/plugins/eslint-plugin-react-dom) for React-specific lint rules:
+
+```js
+// eslint.config.js
+import reactX from 'eslint-plugin-react-x'
+import reactDom from 'eslint-plugin-react-dom'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ // Other configs...
+ // Enable lint rules for React
+ reactX.configs['recommended-typescript'],
+ // Enable lint rules for React DOM
+ reactDom.configs.recommended,
+ ],
+ languageOptions: {
+ parserOptions: {
+ project: ['./tsconfig.node.json', './tsconfig.app.json'],
+ tsconfigRootDir: import.meta.dirname,
+ },
+ // other options...
+ },
+ },
+])
+```

agentflow/web/frontend/eslint.config.js

@@ -0,0 +1,23 @@
+import js from '@eslint/js'
+import globals from 'globals'
+import reactHooks from 'eslint-plugin-react-hooks'
+import reactRefresh from 'eslint-plugin-react-refresh'
+import tseslint from 'typescript-eslint'
+import { defineConfig, globalIgnores } from 'eslint/config'
+
+export default defineConfig([
+ globalIgnores(['dist']),
+ {
+ files: ['**/*.{ts,tsx}'],
+ extends: [
+ js.configs.recommended,
+ tseslint.configs.recommended,
+ reactHooks.configs.flat.recommended,
+ reactRefresh.configs.vite,
+ ],
+ languageOptions: {
+ ecmaVersion: 2020,
+ globals: globals.browser,
+ },
+ },
+])

agentflow/web/frontend/index.html

@@ -0,0 +1,13 @@
+<!doctype html>
+<htmllang="en">
+<head>
+<metacharset="UTF-8" />
+<linkrel="icon" type="image/svg+xml" href="/favicon.svg" />
+<metaname="viewport" content="width=device-width, initial-scale=1.0" />
+<title>frontend</title>
+</head>
+<body>
+<divid="root"></div>
+<scripttype="module" src="/src/main.tsx"></script>
+</body>
+</html>