diff --git a/backend/workflows/__init__.py b/backend/workflows/__init__.py
index c7ffae4c..3ed0654f 100644
--- a/backend/workflows/__init__.py
+++ b/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()
diff --git a/backend/workflows/prose_format_llm/__init__.py b/backend/workflows/prose_format_llm/__init__.py
new file mode 100644
index 00000000..af7c2eb2
--- /dev/null
+++ b/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,
+)
diff --git a/backend/workflows/prose_format_llm/hooks.py b/backend/workflows/prose_format_llm/hooks.py
new file mode 100644
index 00000000..75399ce6
--- /dev/null
+++ b/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"}
diff --git a/backend/workflows/prose_format_llm/loop.py b/backend/workflows/prose_format_llm/loop.py
new file mode 100644
index 00000000..794f86a0
--- /dev/null
+++ b/backend/workflows/prose_format_llm/loop.py
@@ -0,0 +1,229 @@
+"""The judge/enforce loop and the three forced-tool paths.
+
+``run_enforcement_loop`` is the orchestration (pure control flow over injected
+callables). ``make_judge_fn`` / ``make_enforce_fn`` / ``run_analyzer`` are the
+paths that actually call the model via ``forced_tool_call``, shaped per the
+selected prompt mode.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from ..toolkit import forced_tool_call
+from . import TOOL_ANALYZE, TOOL_PATCH, TOOL_REPORT, WORKFLOW_ID
+from .prompts import (
+ ANALYZER_PREAMBLE,
+ ENFORCER_PREAMBLE,
+ JUDGE_PREAMBLE,
+ analyze_instruction,
+ enforce_instruction,
+ judge_instruction,
+ render_spec_block,
+)
+from .violations import clean_analyzer_records, validate_violations
+
+logger = logging.getLogger(__name__)
+
+# Judge classifies, so run it deterministic -- a stable count is what lets the
+# no-progress guard mean something. The enforcer rewrites a span and gets a
+# little headroom.
+_JUDGE_TEMPERATURE = 0.0
+_ENFORCE_TEMPERATURE = 0.2
+_ANALYZE_TEMPERATURE = 0.0
+
+
+def _rail(step: str, i: int, payload: list) -> dict:
+ """A one-line rail summary, emitted regardless of the ``reasoning`` config.
+
+ Shaped as an ``event`` (not a ``type``) so the bridge forwards it to SSE
+ rather than dropping it as an unknown control event. Each delta ends with a
+ newline because the frontend rail concatenates deltas without separators.
+ """
+ pass_id = f"{WORKFLOW_ID}:enforce" if step.startswith("enforce") else f"{WORKFLOW_ID}:judge"
+ if step == "judge":
+ cats = ", ".join(sorted({v["category"] for v in payload}))
+ delta = f"judge iter {i}: {len(payload)} violations" + (f" [{cats}]" if cats else "") + "\n"
+ elif step == "enforce":
+ delta = f"enforce iter {i}: {len(payload)} patches\n"
+ else: # enforce_errors
+ delta = f"enforce iter {i}: {len(payload)} patch(es) skipped\n"
+ return {"event": "reasoning", "data": {"pass": pass_id, "delta": delta}}
+
+
+async def run_enforcement_loop(draft, n, judge_fn, enforce_fn, apply_fn, is_aborted):
+ """Judge -> (enforce -> re-judge)* up to *n* times, yielding rail/reasoning
+ events as it goes and finishing with ``{"type":"loop_done","draft":...}``.
+
+ An async generator, not a coroutine: a workflow hook's only channel to SSE
+ is what it yields, so the final draft rides a terminal sentinel rather than a
+ return value. Stops on a clean judge, no patches, no progress (the count
+ failed to drop -- guards against an enforcer that thrashes), abort, or the
+ cap. ``n == 0`` judges once for diagnosis and never edits.
+ """
+ violations: list = []
+ async for ev in judge_fn(draft):
+ if ev.get("type") == "result":
+ violations = ev["violations"]
+ else:
+ yield ev
+ yield _rail("judge", 0, violations)
+
+ if violations and n > 0:
+ prev = len(violations)
+ for i in range(n):
+ if is_aborted():
+ break
+ patches: list = []
+ async for ev in enforce_fn(draft, violations):
+ if ev.get("type") == "result":
+ patches = ev["patches"]
+ else:
+ yield ev
+ yield _rail("enforce", i + 1, patches)
+ if not patches:
+ break
+ draft, errs = apply_fn(draft, patches)
+ if errs:
+ for e in errs:
+ logger.info("prose_format_llm enforce iter %d: %s", i + 1, e)
+ yield _rail("enforce_errors", i + 1, errs)
+ violations = []
+ async for ev in judge_fn(draft):
+ if ev.get("type") == "result":
+ violations = ev["violations"]
+ else:
+ yield ev
+ yield _rail("judge", i + 1, violations)
+ if not violations or len(violations) >= prev:
+ break
+ prev = len(violations)
+
+ yield {"type": "loop_done", "draft": draft}
+
+
+def _forced(ctx, mode, prefix, tail, tool_name, pass_id, reasoning_on, temperature):
+ """Drive one forced tool call, wired for KV reuse per prompt mode.
+
+ ``extend`` reproduces the pipeline's warm prefix + tools blob; ``minimal``
+ sends its own small self-contained prefix and a single-tool array.
+ """
+ common = dict(
+ client=ctx.client,
+ prefix=prefix,
+ tail_messages=tail,
+ tool_name=tool_name,
+ settings=ctx.settings,
+ pass_id=pass_id,
+ kv_tracker=ctx.kv_tracker,
+ reasoning_on=reasoning_on,
+ temperature=temperature,
+ )
+ if mode == "extend":
+ return forced_tool_call(enabled_tools=ctx.enabled_tools, schema_overrides=ctx.schema_overrides, **common)
+ return forced_tool_call(enabled_tools=None, **common)
+
+
+def _spec_messages(ctx, mode, preamble, spec_block, instruction, draft):
+ """Build (prefix, tail) for a judge/enforce call.
+
+ ``minimal``: preamble + spec ride a fresh system prefix, the draft rides the
+ tail. ``extend``: the draft is replayed as an assistant turn over the
+ pipeline's prefix, and the preamble/spec/instruction ride the final user turn.
+ """
+ if mode == "extend":
+ tail = [
+ {"role": "user", "content": ctx.effective_msg},
+ {"role": "assistant", "content": draft},
+ {
+ "role": "user",
+ "content": f"{preamble}\n\nRecorded prose format:\n{spec_block}\n\n{instruction}\n\n(The draft is the assistant reply directly above.)",
+ },
+ ]
+ return ctx.prefix, tail
+ prefix = [{"role": "system", "content": f"{preamble}\n\nRecorded prose format:\n{spec_block}"}]
+ tail = [{"role": "user", "content": f"{instruction}\n\nDraft:\n{draft}"}]
+ return prefix, tail
+
+
+def make_judge_fn(ctx, spec, mode, reasoning_on, stream):
+ """A factory returning a fresh judge async generator per call (one per loop
+ iteration). ``reasoning_on`` makes the model think; ``stream`` independently
+ decides whether that thinking is surfaced to the rail (via ``pass_id``). Yields
+ the model's reasoning when streamed, then a terminal
+ ``{"type":"result","violations":[...]}`` of validated violations."""
+ spec_block = render_spec_block(spec)
+ filled_keys = list(spec.keys())
+ pass_id = f"{WORKFLOW_ID}:judge" if stream else None
+
+ async def judge_fn(draft):
+ prefix, tail = _spec_messages(ctx, mode, JUDGE_PREAMBLE, spec_block, judge_instruction(TOOL_REPORT), draft)
+ args: dict = {}
+ async for ev in _forced(ctx, mode, prefix, tail, TOOL_REPORT, pass_id, reasoning_on, _JUDGE_TEMPERATURE):
+ if ev.get("type") == "result":
+ args = ev["args"]
+ else:
+ yield ev
+ yield {"type": "result", "violations": validate_violations(args.get("violations"), draft, filled_keys)}
+
+ return judge_fn
+
+
+def make_enforce_fn(ctx, spec, mode, reasoning_on, stream):
+ """A factory returning a fresh enforcer async generator per call.
+ ``reasoning_on`` makes the model think; ``stream`` gates surfacing (``pass_id``).
+ Yields reasoning when streamed, then a terminal ``{"type":"result","patches":[...]}``."""
+ spec_block = render_spec_block(spec)
+ pass_id = f"{WORKFLOW_ID}:enforce" if stream else None
+
+ async def enforce_fn(draft, violations):
+ instruction = enforce_instruction(violations, TOOL_PATCH)
+ prefix, tail = _spec_messages(ctx, mode, ENFORCER_PREAMBLE, spec_block, instruction, draft)
+ args: dict = {}
+ async for ev in _forced(ctx, mode, prefix, tail, TOOL_PATCH, pass_id, reasoning_on, _ENFORCE_TEMPERATURE):
+ if ev.get("type") == "result":
+ args = ev["args"]
+ else:
+ yield ev
+ patches = args.get("patches")
+ yield {"type": "result", "patches": patches if isinstance(patches, list) else []}
+
+ return enforce_fn
+
+
+async def run_analyzer(ctx, schema, *, pass_id, kv_tracker, reasoning_on, client, model_name=None):
+ """Infer the convention from recent prose and yield a terminal
+ ``{"type":"result","values":{...}}``.
+
+ Self-contained (own system prefix, history-derived samples) so the PRE and
+ on-demand paths produce the same prompt -- ``OnDemandCtx`` exposes no pipeline
+ prefix or kv_tracker, and the analyzer runs at most once per conversation, so
+ pipeline cache reuse is moot.
+
+ The lane is the caller's to choose rather than read off ``ctx``: only the
+ off-turn contexts resolve an agent lane, so the in-turn caller has no choice
+ to make and passes its single client with ``model_name=None`` (which falls
+ back to ``settings["model_name"]``).
+ """
+ prefix = [{"role": "system", "content": ANALYZER_PREAMBLE}]
+ tail = [{"role": "user", "content": analyze_instruction(schema, ctx.history, TOOL_ANALYZE)}]
+ args: dict = {}
+ async for ev in forced_tool_call(
+ client=client,
+ prefix=prefix,
+ tail_messages=tail,
+ tool_name=TOOL_ANALYZE,
+ settings=ctx.settings,
+ pass_id=pass_id,
+ enabled_tools=None,
+ kv_tracker=kv_tracker,
+ model_name=model_name,
+ reasoning_on=reasoning_on,
+ temperature=_ANALYZE_TEMPERATURE,
+ ):
+ if ev.get("type") == "result":
+ args = ev["args"]
+ else:
+ yield ev
+ keys = list(schema.keys()) if isinstance(schema, dict) else []
+ yield {"type": "result", "values": clean_analyzer_records(args.get("records"), keys)}
diff --git a/backend/workflows/prose_format_llm/patching.py b/backend/workflows/prose_format_llm/patching.py
new file mode 100644
index 00000000..5a15e9dc
--- /dev/null
+++ b/backend/workflows/prose_format_llm/patching.py
@@ -0,0 +1,47 @@
+"""Pure search/replace patch application.
+
+Deliberately a separate copy of the editor's ``apply_patches`` rather than a
+shared import: the editor's lives in ``backend/pipeline/``, a higher layer this
+workflow may not import. It also omits the editor's quote/asterisk
+normalization fallbacks on purpose -- among other things this workflow fixes
+quote/asterisk markup, so normalizing during the match would hide the very
+markup drift it must catch.
+Keep both facts in mind before "DRY-ing" this away.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def apply_patches(draft: str, patches: Any) -> tuple[str, list[str]]:
+ """Apply each ``{search, replace}`` patch to *draft*, returning the new text
+ and a list of human-readable error strings for patches that were skipped.
+
+ A patch applies only on an exact, unique match; zero or multiple matches are
+ errors (the model must quote more context). Errors are never fatal -- the
+ surviving patches still apply and the caller surfaces the skips.
+ """
+ errors: list[str] = []
+ if not isinstance(patches, list):
+ return draft, errors
+ for i, p in enumerate(patches):
+ if not isinstance(p, dict):
+ errors.append(f"patch {i}: not an object")
+ continue
+ search = p.get("search")
+ replace = p.get("replace")
+ if not isinstance(search, str) or not isinstance(replace, str):
+ errors.append(f"patch {i}: search/replace must be strings")
+ continue
+ if not search or search == replace:
+ continue
+ count = draft.count(search)
+ if count == 0:
+ errors.append(f"patch {i}: search not found")
+ continue
+ if count > 1:
+ errors.append(f"patch {i}: search matched {count} times (not unique)")
+ continue
+ draft = draft.replace(search, replace, 1)
+ return draft, errors
diff --git a/backend/workflows/prose_format_llm/prompts.py b/backend/workflows/prose_format_llm/prompts.py
new file mode 100644
index 00000000..5249bc8a
--- /dev/null
+++ b/backend/workflows/prose_format_llm/prompts.py
@@ -0,0 +1,107 @@
+"""Prompt construction for the analyzer, judge, and enforcer.
+
+Pure string assembly: preambles, the rendered spec/schema/violation blocks, and
+the per-pass task instructions. The message-stack wiring (which block is a
+system message vs a trailing user message, per prompt mode) lives in ``loop.py``.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+# Recent assistant messages the analyzer reads to infer the convention. The
+# format is conventionally stable across a conversation, so a small window is
+# enough and keeps the one-shot analyzer prompt cheap.
+ANALYZER_SAMPLE_WINDOW = 8
+
+ANALYZER_PREAMBLE = (
+ "You are a prose-format analyzer for a roleplay chat. From the conversation's assistant prose, "
+ "you record how each listed element is written, following each element's own guidance -- its "
+ "markup/delimiters, and for narration also its tense (past/present) and narrative person "
+ "(first/second/third). Record only what the prose actually shows."
+)
+
+JUDGE_PREAMBLE = (
+ "You are a prose-format judge for a roleplay chat. Given the recorded prose format and a draft "
+ "reply, you locate whole fragments of the draft that break the recorded format. You only locate "
+ "and categorize -- you never rewrite, suggest fixes, or explain."
+)
+
+ENFORCER_PREAMBLE = (
+ "You are a prose-format enforcer for a roleplay chat. You rewrite each flagged fragment to match "
+ "its recorded format, correcting every deviation at once while keeping the story's events unchanged."
+)
+
+_ANALYZE_TASK = (
+ "For each element below, record how it is written in the prose above, following the element's "
+ "guidance. Skip any element you see no evidence for. Call {tool} with one record per element you "
+ "can characterize.\n\nElements:\n{schema}"
+)
+
+_JUDGE_TASK = (
+ "Find every fragment of the draft that breaks its recorded format. A fragment is one contiguous "
+ "block of a single element (a whole narration passage, a line of dialogue). Report each offending "
+ "fragment ONCE -- the entire fragment as the excerpt, its element name as the category -- even when "
+ "it breaks the format several ways at once (for narration: markup, tense, and narrative person "
+ "together). Never split one fragment into separate entries per issue. Tense and narrative person "
+ "are narration-only; never flag them in dialogue. Do not report compliant fragments. Call {tool}."
+)
+
+_ENFORCE_TASK = (
+ "Each flagged fragment below breaks its recorded format. For each, emit one patch: 'search' is the "
+ "fragment copied verbatim from the draft; 'replace' is that fragment rewritten to its recorded "
+ "format -- fixing markup and, for narration, its tense and narrative person, all together. Changing "
+ "verb tense and pronouns/person is expected; keep every event, detail, and line of dialogue, and "
+ "leave dialogue's own tense and person untouched. Add nothing, drop nothing. Call {tool}."
+ "\n\nFlagged fragments:\n{violations}"
+)
+
+
+def render_spec_block(spec: Mapping[str, str]) -> str:
+ """The recorded format the judge/enforcer hold the draft to, one element per line."""
+ return "\n".join(f"- {k}: {v}" for k, v in spec.items())
+
+
+def render_schema_block(schema: Mapping[str, str]) -> str:
+ """The analyzer's element list: name plus the guidance description for each."""
+ return "\n".join(f"- {k}: {v}" for k, v in schema.items())
+
+
+def render_violations(violations: Sequence[Mapping[str, str]]) -> str:
+ """The judge's findings, formatted for the enforcer: one ``[category] excerpt`` per line."""
+ return "\n".join(f"- [{v['category']}] {v['excerpt']}" for v in violations)
+
+
+def recent_assistant_prose(history: Sequence[Any]) -> str:
+ """Up to ANALYZER_SAMPLE_WINDOW recent assistant messages, oldest-first, joined.
+
+ Assistant history is always plain text; a non-string body (the multimodal
+ list form rides only user messages) carries no prose to sample.
+ """
+ window: list[str] = []
+ for msg in reversed(history):
+ if not isinstance(msg, Mapping) or msg.get("role") != "assistant":
+ continue
+ content = msg.get("content")
+ if isinstance(content, str) and content.strip():
+ window.append(content)
+ if len(window) >= ANALYZER_SAMPLE_WINDOW:
+ break
+ window.reverse()
+ return "\n\n".join(window)
+
+
+def analyze_instruction(schema: Mapping[str, str], history: Sequence[Any], tool: str) -> str:
+ """The analyzer's trailing user message: prose samples plus the element list."""
+ samples = recent_assistant_prose(history) or "(no prior assistant prose)"
+ task = _ANALYZE_TASK.format(tool=tool, schema=render_schema_block(schema))
+ return f"Recent assistant prose:\n{samples}\n\n{task}"
+
+
+def judge_instruction(tool: str) -> str:
+ return _JUDGE_TASK.format(tool=tool)
+
+
+def enforce_instruction(violations: Sequence[Mapping[str, str]], tool: str) -> str:
+ return _ENFORCE_TASK.format(tool=tool, violations=render_violations(violations))
diff --git a/backend/workflows/prose_format_llm/statedoc.py b/backend/workflows/prose_format_llm/statedoc.py
new file mode 100644
index 00000000..f58b4e84
--- /dev/null
+++ b/backend/workflows/prose_format_llm/statedoc.py
@@ -0,0 +1,37 @@
+"""Pure helpers over the per-conversation state document.
+
+State shape: ``{"schema": {elem: description}, "values": {elem: denotation},
+"auto_analyzed": bool}``. ``schema`` guides the analyzer; ``values`` is the only
+thing the judge and enforcer read; ``auto_analyzed`` records that the one
+automatic analysis attempt has fired.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from . import DEFAULT_SCHEMA
+
+
+def seed() -> dict:
+ """A fresh, unarmed state: schema defaults, no recorded values, auto attempt pending."""
+ return {"schema": dict(DEFAULT_SCHEMA), "values": {}, "auto_analyzed": False}
+
+
+def filled_elements(state: Mapping[str, Any] | None) -> dict[str, str]:
+ """The elements the judge/enforcer act on: those with a non-empty string value.
+
+ The ``isinstance`` guard is defensive -- the analyzer and ``save`` only ever
+ write strings, but a malformed LLM reply or a hand-edited slot must not make
+ this (or any caller, e.g. the menu's ``get``) raise.
+ """
+ values = state.get("values") if isinstance(state, Mapping) else None
+ if not isinstance(values, Mapping):
+ return {}
+ return {k: v for k, v in values.items() if isinstance(v, str) and v.strip()}
+
+
+def is_armed(state: Mapping[str, Any] | None) -> bool:
+ """True once at least one element has a recorded value; the loop runs only when armed."""
+ return bool(filled_elements(state))
diff --git a/backend/workflows/prose_format_llm/violations.py b/backend/workflows/prose_format_llm/violations.py
new file mode 100644
index 00000000..52b2fbdc
--- /dev/null
+++ b/backend/workflows/prose_format_llm/violations.py
@@ -0,0 +1,71 @@
+"""Deterministic cleaning of the judge's and analyzer's raw tool output.
+
+Both functions take whatever the model emitted (already coerced to a Python
+value by the tool-call parser, but otherwise untrusted) and return a clean,
+typed result. They are the graceful-degradation boundary for these two tools: a
+malformed reply yields fewer/empty results, never an exception.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+
+
+def validate_violations(raw: Any, draft: str, filled_keys: Iterable[str]) -> list[dict]:
+ """Keep only violations the enforcer can actually act on.
+
+ A violation survives iff its ``excerpt`` is a non-empty string occurring
+ verbatim in *draft* (so a patch can locate it) and its ``category`` is a
+ filled element name (so it maps to a recorded rule). Duplicate
+ ``(excerpt, category)`` pairs collapse to one. The verbatim check doubles as
+ a hallucination filter, which is what makes the surviving count trustworthy
+ enough to drive the loop's no-progress guard.
+ """
+ if not isinstance(raw, list):
+ return []
+ keys = set(filled_keys)
+ seen: set[tuple[str, str]] = set()
+ out: list[dict] = []
+ for v in raw:
+ if not isinstance(v, dict):
+ continue
+ excerpt = v.get("excerpt")
+ category = v.get("category")
+ if not isinstance(excerpt, str) or not excerpt:
+ continue
+ if not isinstance(category, str) or category not in keys:
+ continue
+ if excerpt not in draft:
+ continue
+ key = (excerpt, category)
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append({"excerpt": excerpt, "category": category})
+ return out
+
+
+def clean_analyzer_records(raw: Any, schema_keys: Iterable[str]) -> dict[str, str]:
+ """Map the analyzer's records to ``{element: denotation}``, keeping only
+ schema elements with a non-empty string denotation.
+
+ The string guard is load-bearing: an un-guarded non-string denotation in
+ ``values`` would later crash ``filled_elements``/``is_armed`` (and thus 500
+ the menu's state read) the moment it is consulted.
+ """
+ if not isinstance(raw, list):
+ return {}
+ keys = set(schema_keys)
+ out: dict[str, str] = {}
+ for r in raw:
+ if not isinstance(r, dict):
+ continue
+ category = r.get("category")
+ denotation = r.get("denotation")
+ if not isinstance(category, str) or category not in keys:
+ continue
+ if not isinstance(denotation, str) or not denotation.strip():
+ continue
+ out[category] = denotation
+ return out
diff --git a/frontend/workflows/prose_format_llm/config_panel.js b/frontend/workflows/prose_format_llm/config_panel.js
new file mode 100644
index 00000000..d0c73257
--- /dev/null
+++ b/frontend/workflows/prose_format_llm/config_panel.js
@@ -0,0 +1,232 @@
+// Settings modal for the prose_format_llm workflow. Global behavior config goes
+// through the /config route; the per-conversation prose-format spec goes through
+// the on-demand trigger RPC. Mirrors the TTS panel pattern.
+
+import { api, closeModal, convUrl, esc, getActiveConvId, registerAction, showModal } from "/static/workflow_api.js";
+
+const WORKFLOW_ID = "prose_format_llm";
+
+// One array of {name, description, value} drives both spec sections, so renaming
+// an element never desyncs its recorded value. Rebuilt from the server on every
+// open / analyze / reset.
+let spec = [];
+
+export function initConfigPanel() {
+ registerAction(WORKFLOW_ID, "settings", () => openSettings());
+ registerAction(WORKFLOW_ID, "close", () => closeModal());
+ registerAction(WORKFLOW_ID, "saveGlobal", () => saveGlobal());
+ registerAction(WORKFLOW_ID, "addRow", () => addRow());
+ registerAction(WORKFLOW_ID, "delRow", (el) => delRow(Number(el.dataset.i)));
+ registerAction(WORKFLOW_ID, "analyze", () => analyze());
+ registerAction(WORKFLOW_ID, "saveSpec", () => saveSpec());
+ registerAction(WORKFLOW_ID, "reset", () => reset());
+}
+
+export function configCardRenderer() {
+ return `
Hold replies to a recorded prose format with an LLM judge/enforce pass.
An LLM judge finds where a reply breaks this conversation's recorded format; an enforcer makes the minimal fixes. Dormant until a conversation is analyzed.
+
Enforcement
+
+
+
+
+
+
+
+
+
+
+
+
Loading...
+ `;
+}
+
+async function loadGlobal() {
+ let cfg = {};
+ try {
+ const res = await api.get("/workflows/" + WORKFLOW_ID + "/config");
+ cfg = (res && res.config) || {};
+ } catch (e) {
+ console.warn("prose_format_llm: config load failed", e);
+ }
+ const iters = document.getElementById("pf-cfg-iters");
+ const mode = document.getElementById("pf-cfg-mode");
+ const auto = document.getElementById("pf-cfg-auto");
+ const reasoning = document.getElementById("pf-cfg-reasoning");
+ const stream = document.getElementById("pf-cfg-stream");
+ if (iters) iters.value = Number.isInteger(cfg.max_iterations) ? cfg.max_iterations : 1;
+ if (mode) mode.value = cfg.prompt_mode === "extend" ? "extend" : "minimal";
+ if (auto) auto.checked = !!cfg.auto_analyze;
+ if (reasoning) reasoning.checked = !!cfg.reasoning;
+ if (stream) stream.checked = !!cfg.stream_reasoning;
+}
+
+// The config slot is replaced wholesale on write, so every key must be sent or an
+// omitted one reverts to its default.
+async function saveGlobal() {
+ const iters = parseInt(document.getElementById("pf-cfg-iters")?.value, 10);
+ const config = {
+ max_iterations: Number.isFinite(iters) && iters >= 0 ? iters : 1,
+ prompt_mode: document.getElementById("pf-cfg-mode")?.value === "extend" ? "extend" : "minimal",
+ auto_analyze: !!document.getElementById("pf-cfg-auto")?.checked,
+ reasoning: !!document.getElementById("pf-cfg-reasoning")?.checked,
+ stream_reasoning: !!document.getElementById("pf-cfg-stream")?.checked,
+ };
+ try {
+ await api.put("/workflows/" + WORKFLOW_ID + "/config", { config });
+ } catch (e) {
+ console.warn("prose_format_llm: config save failed", e);
+ }
+}
+
+async function populateSpec() {
+ const host = document.getElementById("pf-spec");
+ if (!host) return;
+ if (!getActiveConvId()) {
+ host.innerHTML = `
Open a conversation to edit its prose format.
`;
+ return;
+ }
+ try {
+ applyState(await postAction({ action: "get" }));
+ } catch (e) {
+ // The trigger route is 404 when the workflow is toggled off.
+ host.innerHTML = `
Enable this workflow to view or edit its prose format.
`;
+ }
+}
+
+function postAction(body) {
+ return api.post(triggerUrl(), body);
+}
+
+function applyState(res) {
+ const schema = (res && res.schema) || {};
+ const values = (res && res.values) || {};
+ const names = Object.keys(schema);
+ for (const k of Object.keys(values)) if (!names.includes(k)) names.push(k);
+ spec = names.map((n) => ({ name: n, description: schema[n] || "", value: values[n] || "" }));
+ renderSpec();
+}
+
+// Read the editable fields back into the spec array before any mutation or save.
+function gather() {
+ for (const el of document.querySelectorAll("[data-pf-name]"))
+ if (spec[+el.dataset.i]) spec[+el.dataset.i].name = el.value;
+ for (const el of document.querySelectorAll("[data-pf-desc]"))
+ if (spec[+el.dataset.i]) spec[+el.dataset.i].description = el.value;
+ for (const el of document.querySelectorAll("[data-pf-val]"))
+ if (spec[+el.dataset.i]) spec[+el.dataset.i].value = el.value;
+}
+
+function renderSpec() {
+ const host = document.getElementById("pf-spec");
+ if (!host) return;
+ const schemaRows = spec
+ .map(
+ (r, i) => `