feat: LLM based prose format conistency workflow

backend/workflows/__init__.py

@@ -58,6 +58,16 @@
 from .image_gen.hooks import regenerate as _image_gen_regenerate
 from .image_gen.hooks import reroll_gen as _image_gen_reroll_gen
 from .image_gen.queries import query as _image_gen_query
+from .prose_format_llm import prose_format_llm_workflow
+from .prose_format_llm.hooks import (
+    on_demand as _pf_on_demand,
+)
+from .prose_format_llm.hooks import (
+    post_pipeline as _pf_post_pipeline,
+)
+from .prose_format_llm.hooks import (
+    pre_pipeline as _pf_pre_pipeline,
+)
 from .registry import (
     Subscription,
     ToolNameCollision,
@@ -161,5 +171,15 @@
 subscribe(image_gen_workflow.id, HookType.REGENERATE, _image_gen_regenerate)
 subscribe(image_gen_workflow.id, HookType.REROLL_GEN, _image_gen_reroll_gen)
 
+# Priority -5 runs the LLM enforcer after the deterministic normalizer (-10) and
+# before any artifact consumer like TTS (0), so an artifact is built from the
+# corrected draft. Ships enabled-but-dormant: nothing runs until a conversation's
+# prose format is analyzed. Replaces format_consistency operationally (disable
+# that one), but the two coexist safely if both stay on.
+register_workflow(prose_format_llm_workflow)
+subscribe(prose_format_llm_workflow.id, HookType.PRE_PIPELINE, _pf_pre_pipeline)
+subscribe(prose_format_llm_workflow.id, HookType.POST_PIPELINE, _pf_post_pipeline, priority=-5)
+subscribe(prose_format_llm_workflow.id, HookType.ON_DEMAND, _pf_on_demand)
+
 
 finalize_registry()

backend/workflows/prose_format_llm/__init__.py

@@ -0,0 +1,143 @@
+"""LLM prose-format workflow.
+
+Enforces a conversation's prose convention -- each element's markup, plus
+narration's tense and narrative person -- on the writer's finished draft, using
+three forced-tool LLM paths: an analyzer that records the convention, a judge
+that locates violations, and an enforcer that patches them. It does the same job as the
+deterministic ``format_consistency`` workflow but covers the open-ended
+violation space regex cannot; deploy it as a replacement by toggling
+``format_consistency`` off.
+
+This module is the data surface only -- constants, the tool schemas, and the
+``Workflow`` record. The hooks, loop, and pure helpers live in sibling modules
+and import these names back; keeping registration out of here avoids an import
+cycle with ``backend/workflows/__init__.py``.
+"""
+
+from __future__ import annotations
+
+from ..contracts import ToolSpec
+from ..registry import Workflow
+
+WORKFLOW_ID = "prose_format_llm"
+
+TOOL_ANALYZE = "prose_format_analyze"
+TOOL_REPORT = "prose_format_report"
+TOOL_PATCH = "prose_format_patch"
+
+# Seeded into each conversation's state. Values are guidance FOR the analyzer --
+# they describe what to record about each element, and are never themselves used
+# to judge a draft. The analyzer overwrites the parallel ``values`` map with the
+# convention it observes; until it does, the conversation is unarmed and the
+# loop stays dormant.
+DEFAULT_SCHEMA = {
+    "narration": (
+        "How narration is written: (1) markup/delimiters (e.g. wrapped in asterisks), "
+        "(2) tense (past or present), and (3) narrative person (first/second/third). "
+        "Tense and person are narration-only -- they never apply to dialogue."
+    ),
+    "speech": "How spoken dialogue is denoted (e.g. text wrapped in double quotes).",
+    "internal_monologue": "How a character's unspoken thought is denoted.",
+    "quotation": "How quoted or cited text inside speech or narration is denoted.",
+}
+
+
+def _array_tool(name: str, description: str, array_key: str, array_description: str, item_props: dict[str, str]) -> ToolSpec:
+    """Build a standalone ToolSpec whose sole parameter is an array of fixed-shape
+    string objects. The schema is static (the per-conversation variation lives in
+    the array's runtime contents, not its shape), so it never busts a cache."""
+    return ToolSpec(
+        name=name,
+        schema={
+            "type": "function",
+            "function": {
+                "name": name,
+                "description": description,
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        array_key: {
+                            "type": "array",
+                            "description": array_description,
+                            "items": {
+                                "type": "object",
+                                "properties": {k: {"type": "string", "description": d} for k, d in item_props.items()},
+                                "required": list(item_props),
+                            },
+                        }
+                    },
+                    "required": [array_key],
+                },
+            },
+        },
+        choice={"type": "function", "function": {"name": name}},
+    )
+
+
+ANALYZE_TOOL = _array_tool(
+    TOOL_ANALYZE,
+    "Record how each prose element is denoted in this conversation.",
+    "records",
+    "One record per element you can characterize from the prose; omit elements with no evidence.",
+    {
+        "category": "The element name, exactly as listed in the request.",
+        "denotation": "How that element is written here: markup, and for narration also its tense and narrative person.",
+    },
+)
+
+REPORT_TOOL = _array_tool(
+    TOOL_REPORT,
+    "Report whole fragments of the draft that violate the recorded prose format.",
+    "violations",
+    "One entry per offending fragment -- the whole fragment, even with several issues; nothing for a clean draft.",
+    {
+        "excerpt": "The whole offending fragment, copied verbatim from the draft.",
+        "category": "The element name the fragment belongs to, exactly as listed.",
+    },
+)
+
+PATCH_TOOL = _array_tool(
+    TOOL_PATCH,
+    "Apply one search/replace edit per flagged fragment to bring it into the recorded format.",
+    "patches",
+    "One patch per flagged fragment.",
+    {
+        "search": "The whole fragment to replace, copied verbatim from the draft.",
+        "replace": "That fragment rewritten to the recorded format (markup, and for narration tense/person); events unchanged.",
+    },
+)
+
+# Every key the hooks read with ``cfg.get(...)`` is present here. The framework
+# returns a non-empty config slot verbatim (no per-key merge with defaults), so a
+# slot must never be written partially; the frontend always sends all keys, and
+# this full default set covers the empty-slot path.
+# ``reasoning`` = the model thinks before each tool call (applies to every agent,
+# every path); ``stream_reasoning`` = surface that thinking on the inspector rail
+# (only the in-turn paths have a stream to surface on). They are independent: a
+# model can reason without the reasoning being shown.
+_CONFIG_DEFAULTS = {
+    "max_iterations": 1,
+    "prompt_mode": "minimal",
+    "auto_analyze": False,
+    "reasoning": False,
+    "stream_reasoning": False,
+}
+
+_CONFIG_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "max_iterations": {"type": "integer", "minimum": 0, "default": 1},
+        "prompt_mode": {"type": "string", "enum": ["minimal", "extend"], "default": "minimal"},
+        "auto_analyze": {"type": "boolean", "default": False},
+        "reasoning": {"type": "boolean", "default": False},
+        "stream_reasoning": {"type": "boolean", "default": False},
+    },
+}
+
+prose_format_llm_workflow = Workflow(
+    id=WORKFLOW_ID,
+    display_name="Prose Format (LLM)",
+    tools=[ANALYZE_TOOL, REPORT_TOOL, PATCH_TOOL],
+    config_defaults=_CONFIG_DEFAULTS,
+    config_schema=_CONFIG_SCHEMA,
+)

backend/workflows/prose_format_llm/hooks.py

@@ -0,0 +1,171 @@
+"""Pipeline and HTTP bindings for the prose-format workflow.
+
+PRE runs the single automatic analysis attempt; POST runs the judge/enforce
+loop over the finished draft; ON_DEMAND is the menu RPC. State reads/writes use
+the toolkit helpers directly: the framework already holds
+``workflow_state_lock`` for the full lifetime of each hook, so they must not
+re-acquire it.
+"""
+
+from __future__ import annotations
+
+from ..contracts import EV_DRAFT_REPLACED
+from ..toolkit import get_workflow_config, get_workflow_state, set_workflow_state
+from . import WORKFLOW_ID
+from .loop import make_enforce_fn, make_judge_fn, run_analyzer, run_enforcement_loop
+from .patching import apply_patches
+from .statedoc import filled_elements, is_armed, seed
+
+
+def _as_n(value) -> int:
+    """Coerce max_iterations to a non-negative int, falling back to the default on
+    a malformed config slot rather than raising on the per-turn path."""
+    try:
+        return max(0, int(value))
+    except (TypeError, ValueError):
+        return 1
+
+
+def _str_map(d: dict) -> dict[str, str]:
+    """Keep only string-valued entries -- a non-string value would later crash the
+    state read that computes ``armed`` (mirrors the analyzer's own guard)."""
+    return {str(k): v for k, v in d.items() if isinstance(v, str)}
+
+
+def _state_payload(state) -> dict:
+    return {"schema": state.get("schema", {}), "values": state.get("values", {}), "armed": is_armed(state)}
+
+
+async def _ensure_seed(conversation_id: str, state):
+    if state is not None:
+        return state
+    state = seed()
+    await set_workflow_state(conversation_id, WORKFLOW_ID, state)
+    return state
+
+
+async def pre_pipeline(ctx):
+    """One automatic analysis attempt per conversation, opt-in via ``auto_analyze``.
+
+    Gated on ``is_armed OR auto_analyzed`` so it fires exactly once when the spec
+    is still empty, then never again automatically -- whether or not that attempt
+    produced anything -- which bounds the per-turn cost. Manual re-analysis is the
+    refresh path.
+    """
+    cfg = await get_workflow_config(WORKFLOW_ID)
+    if not cfg.get("auto_analyze"):
+        return
+    state = await _ensure_seed(ctx.conversation_id, await get_workflow_state(ctx.conversation_id, WORKFLOW_ID))
+    if is_armed(state) or state.get("auto_analyzed"):
+        return
+
+    reasoning_on = bool(cfg.get("reasoning", False))
+    pass_id = f"{WORKFLOW_ID}:analyze" if cfg.get("stream_reasoning") else None
+    values: dict = {}
+    async for ev in run_analyzer(
+        ctx,
+        state.get("schema", {}),
+        pass_id=pass_id,
+        kv_tracker=ctx.kv_tracker,
+        reasoning_on=reasoning_on,
+        client=ctx.client,
+    ):
+        if ev.get("type") == "result":
+            values = ev["values"]
+        else:
+            yield ev
+    merged = {**state, "values": {**state.get("values", {}), **values}, "auto_analyzed": True}
+    await set_workflow_state(ctx.conversation_id, WORKFLOW_ID, merged)
+
+
+async def post_pipeline(ctx):
+    """Run the judge/enforce loop on the draft when the spec is armed.
+
+    Dormant otherwise (no LLM call). The phase pill is yielded here, around the
+    loop, so it clears even when the loop made no edit; the loop's own events are
+    re-yielded and its final draft is read off the terminal sentinel.
+    """
+    state = await get_workflow_state(ctx.conversation_id, WORKFLOW_ID)
+    if state is None or not is_armed(state):
+        return
+
+    cfg = await get_workflow_config(WORKFLOW_ID)
+    n = _as_n(cfg.get("max_iterations", 1))
+    mode = cfg.get("prompt_mode") or "minimal"
+    reasoning_on = bool(cfg.get("reasoning", False))
+    stream = bool(cfg.get("stream_reasoning", False))
+    spec = filled_elements(state)
+
+    judge_fn = make_judge_fn(ctx, spec, mode, reasoning_on, stream)
+    enforce_fn = make_enforce_fn(ctx, spec, mode, reasoning_on, stream)
+
+    def is_aborted():
+        return ctx.client.is_aborted
+
+    yield {"event": "phase_status", "data": {"channel": f"workflow:{WORKFLOW_ID}", "label": "Enforcing format"}}
+    final = ctx.draft
+    async for ev in run_enforcement_loop(ctx.draft, n, judge_fn, enforce_fn, apply_patches, is_aborted):
+        if ev.get("type") == "loop_done":
+            final = ev["draft"]
+        else:
+            yield ev
+    if final != ctx.draft:
+        yield {"type": EV_DRAFT_REPLACED, "draft": final}
+    yield {"event": "phase_status", "data": {"channel": f"workflow:{WORKFLOW_ID}", "state": "done"}}
+
+
+async def on_demand(ctx, body):
+    """Menu RPC dispatched on ``body['action']``. Seeds state on first contact so
+    the menu always has a schema to show."""
+    action = body.get("action") if isinstance(body, dict) else None
+    state = await _ensure_seed(ctx.conversation_id, await get_workflow_state(ctx.conversation_id, WORKFLOW_ID))
+
+    if action == "get":
+        return _state_payload(state)
+
+    if action == "save":
+        new = dict(state)
+        if isinstance(body.get("schema"), dict):
+            new["schema"] = _str_map(body["schema"])
+        if isinstance(body.get("values"), dict):
+            new["values"] = _str_map(body["values"])
+        await set_workflow_state(ctx.conversation_id, WORKFLOW_ID, new)
+        return _state_payload(new)
+
+    if action == "analyze":
+        # Reasoning (the "think" knob) applies to every agent, this one included.
+        # On-demand has no SSE stream, so the reasoning is never surfaced here
+        # regardless of stream_reasoning (pass_id stays None) -- the model still
+        # thinks when reasoning is on. auto_analyzed is left untouched: this is the
+        # manual refresh, not the one auto attempt.
+        #
+        # Off-turn tool-calling work runs on the agent lane, matching the other
+        # off-turn workflow hooks. In a dual-model setup that puts this analyzer on
+        # a different model than the pre-pipeline one, which reads the same prose
+        # and writes the same slot: PreCtx carries no agent lane, so the automatic
+        # attempt cannot follow. The two can therefore record different conventions
+        # for one conversation.
+        cfg = await get_workflow_config(WORKFLOW_ID)
+        reasoning_on = bool(cfg.get("reasoning", False))
+        values: dict = {}
+        async for ev in run_analyzer(
+            ctx,
+            state.get("schema", {}),
+            pass_id=None,
+            kv_tracker=None,
+            reasoning_on=reasoning_on,
+            client=ctx.agent_client,
+            model_name=ctx.agent_model_name,
+        ):
+            if ev.get("type") == "result":
+                values = ev["values"]
+        merged = {**state, "values": {**state.get("values", {}), **values}}
+        await set_workflow_state(ctx.conversation_id, WORKFLOW_ID, merged)
+        return _state_payload(merged)
+
+    if action == "reset":
+        new = seed()
+        await set_workflow_state(ctx.conversation_id, WORKFLOW_ID, new)
+        return _state_payload(new)
+
+    return {"error": "unknown action"}