Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 81 additions & 60 deletions src/spark_character/output_sanitizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,68 +42,89 @@ def strip_format_controls(text: str) -> str:


def replace_em_dashes(text: str, replacement: str = " - ") -> str:
"""Replace Unicode dash punctuation with a plain hyphen separator.

Default replacement is " - " (space-hyphen-space) to match the
typographic role an em dash usually plays as a parenthetical
separator. The function then collapses any double spaces this
introduces, so existing single-spaced "word — word" becomes
"word - word", not "word - word".
"""
if not text:
return text
out = unicodedata.normalize("NFKD", text)
out = "".join(replacement if is_dash_punctuation(ch) else ch for ch in out)
while " " in out:
out = out.replace(" ", " ")
return out


if not isinstance(text, str): text = str(text or '')
if not isinstance(replacement, str): replacement = str(replacement or '')
try:
"""Replace Unicode dash punctuation with a plain hyphen separator.

Default replacement is " - " (space-hyphen-space) to match the
typographic role an em dash usually plays as a parenthetical
separator. The function then collapses any double spaces this
introduces, so existing single-spaced "word — word" becomes
"word - word", not "word - word".
"""
if not text:
return text
out = unicodedata.normalize("NFKD", text)
out = "".join(replacement if is_dash_punctuation(ch) else ch for ch in out)
while " " in out:
out = out.replace(" ", " ")
return out



except Exception:
return ""
def strip_markdown_emphasis(text: str) -> str:
"""Remove paired bold/italic emphasis markers while preserving bullets."""
if not text:
return text
out = re.sub(r"\*\*\*([^*\n][\s\S]*?[^*\n])\*\*\*", r"\1", text)
out = re.sub(r"\*\*([^*\n][\s\S]*?[^*\n])\*\*", r"\1", out)
out = re.sub(r"__([^_\n][\s\S]*?[^_\n])__", r"\1", out)
return out
if not isinstance(text, str): text = str(text or '')
try:
"""Remove paired bold/italic emphasis markers while preserving bullets."""
if not text:
return text
out = re.sub(r"\*\*\*([^*\n][\s\S]*?[^*\n])\*\*\*", r"\1", text)
out = re.sub(r"\*\*([^*\n][\s\S]*?[^*\n])\*\*", r"\1", out)
out = re.sub(r"__([^_\n][\s\S]*?[^_\n])__", r"\1", out)
return out


def rewrite_spawner_surface_standalone_question(text: str) -> str:
"""Replace stale standalone-tool questions for existing Spawner surfaces."""
if not text:
return text
lower = text.lower()
mentions_spawner_surface = "spawner" in lower and (
"kanban" in lower or "canvas" in lower or "mission control" in lower
)
if not mentions_spawner_surface or "standalone" not in lower:
return text

replacement = (
"\n\nSince this lives inside the existing Spawner UI routes, the useful question is: "
"which surface should we tighten first - Kanban state accuracy, Canvas execution state, "
"or Telegram relay messaging?"
)
out = re.sub(
r"\n?\s*[-*]?\s*Are you thinking this (?:runs locally as|should run locally as|should be) "
r"a standalone (?:page|app|tool),\s*or lives? inside the existing Spawner UI routes\?\s*$",
replacement,
text,
flags=re.IGNORECASE,
)
out = re.sub(
r"\n?\s*[-*]?\s*Should this be a standalone (?:page|app|tool),\s*"
r"or live inside the existing Spawner UI routes\?\s*$",
replacement,
out,
flags=re.IGNORECASE,
)
return out


except Exception:
return ""
def rewrite_spawner_surface_standalone_question(text: str) -> str:
if not isinstance(text, str): text = str(text or '')
try:
"""Replace stale standalone-tool questions for existing Spawner surfaces."""
if not text:
return text
lower = text.lower()
mentions_spawner_surface = "spawner" in lower and (
"kanban" in lower or "canvas" in lower or "mission control" in lower
)
if not mentions_spawner_surface or "standalone" not in lower:
return text

replacement = (
"\n\nSince this lives inside the existing Spawner UI routes, the useful question is: "
"which surface should we tighten first - Kanban state accuracy, Canvas execution state, "
"or Telegram relay messaging?"
)
out = re.sub(
r"\n?\s*[-*]?\s*Are you thinking this (?:runs locally as|should run locally as|should be) "
r"a standalone (?:page|app|tool),\s*or lives? inside the existing Spawner UI routes\?\s*$",
replacement,
text,
flags=re.IGNORECASE,
)
out = re.sub(
r"\n?\s*[-*]?\s*Should this be a standalone (?:page|app|tool),\s*"
r"or live inside the existing Spawner UI routes\?\s*$",
replacement,
out,
flags=re.IGNORECASE,
)
return out



except Exception:
return ""
def sanitize_voice_output(text: str) -> str:
"""Apply all voice post-processors that are safe to run in production."""
return rewrite_spawner_surface_standalone_question(
strip_markdown_emphasis(replace_em_dashes(strip_format_controls(text)))
)
if not isinstance(text, str): text = str(text or '')
try:
"""Apply all voice post-processors that are safe to run in production."""
return rewrite_spawner_surface_standalone_question(
strip_markdown_emphasis(replace_em_dashes(strip_format_controls(text)))
)

except Exception:
return ""
23 changes: 14 additions & 9 deletions src/spark_character/persona.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,17 +35,22 @@ def system_prompt(self) -> str:


def load_overlay(provider_kind: str | None) -> str:
"""Return the overlay markdown for a given backend kind, or '' if
none is configured. Provider kinds: 'zai', 'minimax', 'codex',
'openai', 'ollama'. Unknown kinds return ''."""
if not provider_kind:
return ""
path = OVERLAYS_DIR / f"{provider_kind.lower().strip()}.md"
if not path.exists():
return ""
return sanitize_prompt_text(path.read_text(encoding="utf-8")).strip()
if not isinstance(provider_kind, str): provider_kind = str(provider_kind or '')
try:
"""Return the overlay markdown for a given backend kind, or '' if
none is configured. Provider kinds: 'zai', 'minimax', 'codex',
'openai', 'ollama'. Unknown kinds return ''."""
if not provider_kind:
return ""
path = OVERLAYS_DIR / f"{provider_kind.lower().strip()}.md"
if not path.exists():
return ""
return sanitize_prompt_text(path.read_text(encoding="utf-8")).strip()



except Exception:
return ""
def load_surface_overlay(surface: str | None) -> str:
"""Return the overlay markdown for a given surface, or '' if none
is configured. Surfaces: 'voice', 'browser_extension', 'telegram',
Expand Down