feat(flows): agent-friendliness — editing, drafts, safety rails, wider tools

app/src/hooks/useFlowChanged.ts

@@ -0,0 +1,85 @@
+/**
+ * useFlowChanged (Phase 3 — flow-mutation observability, audit F6)
+ * ----------------------------------------------------------------
+ *
+ * Subscribes to the core's flow-mutation feed so an open Workflows list or
+ * canvas reacts when a flow's definition changes underneath it — most
+ * importantly, so an agent `save_workflow` becomes visible instead of the user
+ * silently working against (and later clobbering) stale state.
+ *
+ * The backend publishes `DomainEvent::FlowChanged` on create/update/delete/
+ * enable; the core socket bridge (`src/core/socketio.rs`) re-emits it as both
+ * `flow:changed` and `flow_changed` (colon + underscore aliases) with the
+ * payload `{ flow_id, kind, actor }`.
+ *
+ * Best-effort (broadcast bridges drop on lag); the UI's own refetch-on-focus
+ * remains the backstop, exactly as {@link useFlowRunProgress} pairs with the 2s
+ * poller. Pass `flowId` to filter to a single flow (canvas), or omit it to
+ * receive every change (list).
+ */
+import debug from 'debug';
+import { useCallback, useEffect } from 'react';
+
+import { socketService } from '../services/socketService';
+
+const log = debug('flows:changed');
+
+/** Socket event aliases the core bridge emits (colon + underscore forms). */
+const EVENT_COLON = 'flow:changed';
+const EVENT_UNDERSCORE = 'flow_changed';
+
+/** What happened to the flow. */
+export type FlowChangeKind = 'created' | 'updated' | 'deleted' | 'enabled_changed' | (string & {});
+
+/** Payload of a `flow:changed` socket event (`DomainEvent::FlowChanged`). */
+export interface FlowChangedEvent {
+  flow_id: string;
+  kind: FlowChangeKind;
+  /** Coarse hint: `agent` | `user` | `system`. */
+  actor: string;
+}
+
+function parsePayload(data: unknown): FlowChangedEvent | null {
+  if (!data || typeof data !== 'object') return null;
+  const obj = data as Record<string, unknown>;
+  if (typeof obj.flow_id !== 'string' || typeof obj.kind !== 'string') return null;
+  return {
+    flow_id: obj.flow_id,
+    kind: obj.kind,
+    actor: typeof obj.actor === 'string' ? obj.actor : 'system',
+  };
+}
+
+/**
+ * Invokes `onChange` whenever a flow changes. When `flowId` is provided, only
+ * changes to that flow are delivered; otherwise every change is.
+ */
+export function useFlowChanged(
+  onChange: (event: FlowChangedEvent) => void,
+  flowId?: string | null
+): void {
+  const handle = useCallback(
+    (data: unknown) => {
+      const payload = parsePayload(data);
+      if (!payload) {
+        log('changed: dropped — invalid payload %o', data);
+        return;
+      }
+      if (flowId && payload.flow_id !== flowId) return;
+      log('changed: flow=%s kind=%s actor=%s', payload.flow_id, payload.kind, payload.actor);
+      onChange(payload);
+    },
+    [onChange, flowId]
+  );
+
+  useEffect(() => {
+    socketService.on(EVENT_COLON, handle);
+    socketService.on(EVENT_UNDERSCORE, handle);
+    return () => {
+      socketService.off(EVENT_COLON, handle);
+      socketService.off(EVENT_UNDERSCORE, handle);
+    };
+  }, [handle]);
+}
+
+export default useFlowChanged;

app/src/pages/FlowsPage.tsx

@@ -31,6 +31,7 @@ import BetaBanner from '../components/ui/BetaBanner';
 import Button from '../components/ui/Button';
 import { CenteredLoadingState, ErrorBanner } from '../components/ui/LoadingState';
 import { ModalShell } from '../components/ui/ModalShell';
+import { useFlowChanged } from '../hooks/useFlowChanged';
 import { FLOW_CANVAS_DRAFT_ROUTE, type FlowCanvasDraftState } from '../lib/flows/canvasDraft';
 import { downloadFlowGraph } from '../lib/flows/exportFlow';
 import { type FlowTemplate, templateNameKey } from '../lib/flows/templates';
@@ -105,6 +106,17 @@ export default function FlowsPage() {
     void loadFlows();
   }, [loadFlows]);
 
+  // Refetch (silently, no spinner) whenever any flow changes underneath us —
+  // e.g. an agent save_workflow — so the list never shows stale state (F6).
+  useFlowChanged(
+    useCallback(() => {
+      log('flow:changed — refetching list');
+      void listFlows()
+        .then(setFlows)
+        .catch(err => log('refetch failed: %o', err));
+    }, [])
+  );
+
   const handleToggle = useCallback(
     async (flow: Flow) => {
       if (busyKey) return;

app/src/services/api/flowsApi.ts

@@ -142,10 +142,45 @@ export interface Flow {
  */
 export interface FlowValidation {
   valid: boolean;
+  /** All structural errors in one pass (multi-error validation), not just the first. */
   errors: string[];
+  /** Structured, per-node counterpart to {@link errors} (additive). */
+  error_details?: FlowValidationErrorDetail[];
   warnings: string[];
 }
 
+/** One structured validation error (`src/openhuman/flows/types.rs::FlowValidationError`). */
+export interface FlowValidationErrorDetail {
+  /** Stable machine-readable code, e.g. `missing_trigger`, `unknown_node`. */
+  code: string;
+  /** Human-readable message (identical to the matching {@link FlowValidation.errors} entry). */
+  message: string;
+  /** The node this error anchors to, when node-specific. */
+  node_id?: string;
+  /** The offending config field, when field-specific (reserved). */
+  field?: string;
+}
+
+/** Where a {@link FlowDraft} originated (`src/openhuman/flows/types.rs::DraftOrigin`). */
+export type DraftOrigin = 'chat' | 'canvas' | 'import';
+
+/**
+ * A core-managed, durable workflow draft (`src/openhuman/flows/types.rs::FlowDraft`)
+ * — the shared working copy the agent tools and the canvas both read/write by
+ * id across turns and reloads. Never live; promote runs the normal save gates.
+ */
+export interface FlowDraft {
+  id: string;
+  /** The saved flow this draft edits, if any (promote → update vs create). */
+  flow_id?: string;
+  name: string;
+  /** Work-in-progress graph (may be incomplete/invalid) — opaque to this client. */
+  graph: unknown;
+  origin: DraftOrigin;
+  created_at: string;
+  updated_at: string;
+}
+
 /**
  * Source format for {@link importFlow}. `native` is a tinyflows `WorkflowGraph`
  * JSON; `n8n` is an n8n workflow export (mapped best-effort host-side); `auto`
@@ -185,6 +220,54 @@ export interface FlowUpdate {
   name?: string;
   graph?: unknown;
   requireApproval?: boolean;
+  /**
+   * Optimistic-concurrency token: the flow's `updated_at` as last observed. If
+   * the flow changed since, the update is refused with a {@link FlowVersionConflict}
+   * error instead of clobbering. Omit for last-write-wins.
+   */
+  expectedVersion?: string;
+  /** Run the agent author hard-gates before persisting (F3). */
+  strict?: boolean;
+}
+
+/** A revision snapshot (`src/openhuman/flows/types.rs::FlowRevision`). */
+export interface FlowRevision {
+  id: string;
+  flow_id: string;
+  graph: unknown;
+  name: string;
+  require_approval: boolean;
+  created_at: string;
+}
+
+/**
+ * The structured error `flows_update` returns on an optimistic-concurrency
+ * conflict (encoded in the RPC error message as JSON). Detect it by parsing a
+ * caught update error — see {@link parseFlowVersionConflict}.
+ */
+export interface FlowVersionConflict {
+  code: 'version_conflict';
+  message: string;
+  current: Flow;
+}
+
+/**
+ * If `err` is a `flows_update` version-conflict error, returns the structured
+ * conflict (with the current server flow) so the UI can offer reload/diff;
+ * otherwise `null`.
+ */
+export function parseFlowVersionConflict(err: unknown): FlowVersionConflict | null {
+  const message = err instanceof Error ? err.message : typeof err === 'string' ? err : '';
+  if (!message.includes('version_conflict')) return null;
+  try {
+    const parsed = JSON.parse(message) as Partial<FlowVersionConflict>;
+    if (parsed?.code === 'version_conflict' && parsed.current) {
+      return parsed as FlowVersionConflict;
+    }
+  } catch {
+    // Not a JSON conflict payload.
+  }
+  return null;
 }
 
 /** Lifecycle status of a {@link FlowSuggestion} (`src/openhuman/flows/types.rs::SuggestionStatus`). */
@@ -464,12 +547,42 @@ export async function updateFlow(id: string, update: FlowUpdate): Promise<Flow>
   if (update.name !== undefined) params.name = update.name;
   if (update.graph !== undefined) params.graph = update.graph;
   if (update.requireApproval !== undefined) params.require_approval = update.requireApproval;
+  if (update.expectedVersion !== undefined) params.expected_version = update.expectedVersion;
+  if (update.strict !== undefined) params.strict = update.strict;
   const response = await callCoreRpc<unknown>({ method: 'openhuman.flows_update', params });
   const flow = unwrapCliEnvelope<Flow>(response);
   log('updateFlow: response id=%s name=%s', flow.id, flow.name);
   return flow;
 }
 
+/** List a flow's revision history via `openhuman.flows_get_history` (newest first). */
+export async function getFlowHistory(id: string, limit?: number): Promise<FlowRevision[]> {
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_get_history',
+    params: { id, limit },
+  });
+  const result = unwrapCliEnvelope<{ revisions: FlowRevision[] }>(response);
+  return result.revisions ?? [];
+}
+
+/**
+ * Roll a flow back to a prior revision via `openhuman.flows_rollback` (restores
+ * that revision's graph through the normal update path — itself snapshotted, so
+ * a rollback is undoable). Honours optimistic concurrency via `expectedVersion`.
+ */
+export async function rollbackFlow(
+  id: string,
+  revisionId: string,
+  expectedVersion?: string
+): Promise<Flow> {
+  log('rollbackFlow: request id=%s revision=%s', id, revisionId);
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_rollback',
+    params: { id, revision_id: revisionId, expected_version: expectedVersion },
+  });
+  return unwrapCliEnvelope<Flow>(response);
+}
+
 /**
  * Validate a candidate `WorkflowGraph` via `openhuman.flows_validate`. Pure and
  * cheap server-side (no config load), so it's safe to call on a debounce while
@@ -532,6 +645,139 @@ export async function importFlow(
   return result;
 }
 
+// ── Catalog RPCs for the UI (Phase 5, item 16) ───────────────────────────────
+
+/** One search hit from `openhuman.flows_search_tool_catalog` (secret-free). */
+export interface ToolCatalogEntry {
+  slug: string;
+  toolkit: string;
+  description?: string | null;
+  required_args?: string[];
+  output_fields?: string[];
+  primary_array_path?: string | null;
+  /** Curated/featured toolkits rank first. */
+  featured?: boolean;
+}
+
+/** Search the live Composio tool catalog via `openhuman.flows_search_tool_catalog`. */
+export async function searchToolCatalog(
+  query: string,
+  opts?: { toolkit?: string; limit?: number }
+): Promise<ToolCatalogEntry[]> {
+  log('searchToolCatalog: query=%s toolkit=%s', query, opts?.toolkit ?? '(all)');
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_search_tool_catalog',
+    params: { query, toolkit: opts?.toolkit, limit: opts?.limit },
+    timeoutMs: 60_000,
+  });
+  const result = unwrapCliEnvelope<{ tools: ToolCatalogEntry[] }>(response);
+  return result.tools ?? [];
+}
+
+/** A toolkit a graph needs, with its connected state (Phase 5, item 18). */
+export interface RequiredConnection {
+  toolkit: string;
+  status: 'connected' | 'missing';
+}
+
+/**
+ * Compute which Composio toolkits a candidate graph needs and whether each is
+ * connected, via `openhuman.flows_required_connections` — the data behind the
+ * "Connect <toolkit>" CTAs. Also surfaced on the workflow_proposal payload.
+ */
+export async function requiredConnections(graph: unknown): Promise<RequiredConnection[]> {
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_required_connections',
+    params: { graph },
+  });
+  const result = unwrapCliEnvelope<{ required_connections: RequiredConnection[] }>(response);
+  return result.required_connections ?? [];
+}
+
+/** Fetch one action's full contract via `openhuman.flows_get_tool_contract`. */
+export async function getToolContract(slug: string): Promise<unknown> {
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_get_tool_contract',
+    params: { slug },
+    timeoutMs: 60_000,
+  });
+  const result = unwrapCliEnvelope<{ contract: unknown }>(response);
+  return result.contract;
+}
+
+// ── Core-managed drafts (F5) ─────────────────────────────────────────────────
+
+/** Create a durable draft via `openhuman.flows_draft_create`. */
+export async function createDraft(params: {
+  name: string;
+  graph: unknown;
+  flowId?: string;
+  origin?: DraftOrigin;
+}): Promise<FlowDraft> {
+  log('createDraft: request origin=%s', params.origin ?? 'canvas');
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_draft_create',
+    params: {
+      name: params.name,
+      graph: params.graph,
+      flow_id: params.flowId,
+      origin: params.origin,
+    },
+  });
+  return unwrapCliEnvelope<FlowDraft>(response);
+}
+
+/** Fetch a draft by id via `openhuman.flows_draft_get`. */
+export async function getDraft(id: string): Promise<FlowDraft> {
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_draft_get',
+    params: { id },
+  });
+  return unwrapCliEnvelope<FlowDraft>(response);
+}
+
+/** Patch a draft's name/graph/flow_id via `openhuman.flows_draft_update`. */
+export async function updateDraft(
+  id: string,
+  patch: { name?: string; graph?: unknown; flowId?: string }
+): Promise<FlowDraft> {
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_draft_update',
+    params: { id, name: patch.name, graph: patch.graph, flow_id: patch.flowId },
+  });
+  return unwrapCliEnvelope<FlowDraft>(response);
+}
+
+/** List all drafts (newest-updated first) via `openhuman.flows_draft_list`. */
+export async function listDrafts(): Promise<FlowDraft[]> {
+  const response = await callCoreRpc<unknown>({ method: 'openhuman.flows_draft_list', params: {} });
+  const result = unwrapCliEnvelope<{ drafts: FlowDraft[] }>(response);
+  return result.drafts ?? [];
+}
+
+/** Delete a draft via `openhuman.flows_draft_delete`. */
+export async function deleteDraft(id: string): Promise<boolean> {
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_draft_delete',
+    params: { id },
+  });
+  const result = unwrapCliEnvelope<{ id: string; deleted: boolean }>(response);
+  return result.deleted;
+}
+
+/**
+ * Promote a draft into a saved flow via `openhuman.flows_draft_promote` (runs
+ * the normal create/update gates, then removes the draft). Returns the Flow.
+ */
+export async function promoteDraft(id: string, requireApproval?: boolean): Promise<Flow> {
+  log('promoteDraft: request id=%s', id);
+  const response = await callCoreRpc<unknown>({
+    method: 'openhuman.flows_draft_promote',
+    params: { id, require_approval: requireApproval },
+  });
+  return unwrapCliEnvelope<Flow>(response);
+}
+
 /**
  * `openhuman.flows_discover` runs the read-only Flow Scout agent, which reasons
  * over the user's memory/threads/connections/flows and can take up to ~300s