fix: address self-review findings

This commit is contained in:
Matt Van Horn
2026-08-11 19:25:40 -07:00
parent 410e2f6627
commit 9e50da6d67
2 changed files with 107 additions and 16 deletions
+65 -2
View File
@@ -159,6 +159,9 @@ A --- B
B --x C
C --o D
D ----> E
E -.- F
F === G
A --> B; B --> C
""",
encoding="utf-8",
)
@@ -169,6 +172,15 @@ D ----> E
fail("cross/circle arrowheads were not normalized")
if edges[3]["style"] != "solid":
fail("long edge form did not discard length while retaining style")
if edges[4]["style"] != "dashed" or not edges[4]["undirected"]:
fail("open dotted edge was not retained as dashed and undirected")
if edges[5]["style"] != "thick" or not edges[5]["undirected"]:
fail("open thick edge was not retained as thick and undirected")
if [(edge["source"], edge["target"]) for edge in edges[6:]] != [
("A", "B"),
("B", "C"),
]:
fail("semicolon-separated flowchart statements were not split")
ok("documented shape families and edge forms normalize")
@@ -195,6 +207,37 @@ def check_markdown_and_grammars(tmp: Path) -> None:
if not sequence["notes"]:
fail("sequence note was not retained as inert text")
fragments = tmp / "sequence-fragments.mmd"
fragments.write_text(
"""sequenceDiagram
participant A
participant B
alt outer
critical must succeed
A->>B: try
option fallback
A->>B: recover
end
break stop now
B-->>A: stop
end
else continue
A->>B: continue
end
""",
encoding="utf-8",
)
fragment_payload = json.loads(run_extract([str(fragments), "--json"]))["diagrams"][0]
fragment_entries = fragment_payload["fragments"]
if [entry["kind"] for entry in fragment_entries] != ["alt", "critical", "break"]:
fail("critical and break sequence fragments were not retained")
if fragment_entries[1]["regions"] != ["fallback"]:
fail("critical option region was not retained")
if [entry["depth"] for entry in fragment_entries] != [0, 1, 1]:
fail("sequence fragment nesting was not retained")
if fragment_entries[0]["regions"] != ["continue"]:
fail("nested fragment end popped the wrong sequence fragment")
state = tmp / "states.mmd"
state.write_text(
"""stateDiagram-v2
@@ -206,6 +249,8 @@ state Running {
Idle --> Running: start
Running --> [*]: stop
state fork_join <<fork>>
s1 : Waiting for work
s1 --> Idle: ready
""",
encoding="utf-8",
)
@@ -218,6 +263,9 @@ state fork_join <<fork>>
fail("state start marker was not exposed as an entry point")
if not state_payload["analysis"]["terminals"]:
fail("state end marker was not exposed as a terminal")
state_nodes = {node["id"]: node for node in state_payload["nodes"]}
if state_nodes["s1"]["label"] != "Waiting for work":
fail("colon-form state description was not retained")
er = tmp / "model.mermaid"
er.write_text(
@@ -242,7 +290,7 @@ CUSTOMER ||--o{ ORDER : places
ok("Markdown selection plus sequence, state, and ER grammars parse")
def check_adversarial() -> None:
def check_adversarial(tmp: Path) -> None:
payload = json.loads(run_extract([str(ADVERSARIAL), "--json"]))
diagram = payload["diagrams"][0]
nodes = {node["id"]: node for node in diagram["nodes"]}
@@ -271,6 +319,21 @@ def check_adversarial() -> None:
fail("click URL crossed the trust boundary into output")
if "IGNORE ALL PREVIOUS INSTRUCTIONS" not in serialized:
fail("prompt-injection label was not retained as inert diagram text")
markdown_payload = tmp / "markdown-payload.mmd"
markdown_payload.write_text(
'flowchart TD\nA["&lt;img src=https://example.invalid/pixel&gt; '
'![remote](https://example.invalid/image) [link](https://example.invalid)"]\n',
encoding="utf-8",
)
digest_path = tmp / "digest.md"
run_extract([str(markdown_payload), "--out", str(digest_path)])
digest_text = digest_path.read_text(encoding="utf-8")
for active in ("<img", "![remote]", "[link](https://example.invalid)"):
if active in digest_text:
fail(f"Markdown digest reactivated untrusted label content: {active!r}")
if "&lt;img src=" not in digest_text or "&gt;" not in digest_text:
fail("decoded HTML label was not safely encoded in Markdown output")
ok("adversarial labels stay inert; nesting, chains, fan-out, discards work")
@@ -418,7 +481,7 @@ def main() -> int:
check_flowchart()
check_shape_and_edge_vocabulary(tmp)
check_markdown_and_grammars(tmp)
check_adversarial()
check_adversarial(tmp)
check_errors_and_limits(tmp)
check_docs_and_wiring()
print("\nAll Mermaid import gates passed.")
@@ -405,7 +405,7 @@ class _Operator:
def _operator_style(token: str) -> tuple[str, str, bool, bool]:
style = "dashed" if "." in token else "thick" if "=" in token else "solid"
arrowhead = "cross" if token.endswith("x") else "circle" if token.endswith("o") else "arrow"
undirected = token.endswith("-") and ">" not in token
undirected = ">" not in token and not token.endswith(("x", "o"))
bidirectional = token.startswith("<") and token.endswith(">")
return style, arrowhead, bidirectional, undirected
@@ -432,7 +432,7 @@ def _edge_operators(text: str) -> list[_Operator]:
)
occupied.append((match.start(), match.end()))
pattern = re.compile(r"<[-=.]+>|-+\.-+>|=+>|-+(?:>|x|o)|-{3,}")
pattern = re.compile(r"<[-=.]+>|-+\.-+>|=+>|-+(?:>|x|o)|-+\.-+|={3,}|-{3,}")
for match in pattern.finditer(mask):
if any(start <= match.start() < end for start, end in occupied):
continue
@@ -485,7 +485,12 @@ def _parse_flowchart(
diagram: Diagram, lines: list[tuple[int, str]], header_position: int
) -> None:
containers: list[str] = []
for line_number, raw in lines[header_position + 1 :]:
statements = (
(line_number, statement)
for line_number, raw in lines[header_position + 1 :]
for statement in _split_top_level(raw, ";")
)
for line_number, raw in statements:
text = raw.strip()
if not text:
continue
@@ -577,7 +582,7 @@ def _parse_sequence(
"actor" if participant.group(1).casefold() == "actor" else "lifeline",
)
continue
fragment = re.match(r"^(alt|opt|loop|par)\b\s*(.*)$", text, re.I)
fragment = re.match(r"^(alt|opt|loop|par|critical|break)\b\s*(.*)$", text, re.I)
if fragment:
entry = {
"kind": fragment.group(1).casefold(),
@@ -589,7 +594,7 @@ def _parse_sequence(
diagram.fragments.append(entry)
fragment_stack.append(entry)
continue
region = re.match(r"^(else|and)\b\s*(.*)$", text, re.I)
region = re.match(r"^(else|and|option)\b\s*(.*)$", text, re.I)
if region and fragment_stack:
fragment_stack[-1]["regions"].append(clean_label(region.group(2)))
continue
@@ -688,6 +693,12 @@ def _parse_state(
continue
if "-->" in text:
_fail(f"malformed edge at line {line_number}")
description = re.match(r"^([A-Za-z_][\w.-]*)\s*:\s*(.+)$", text)
if description:
diagram.add_node(
description.group(1), clean_label(description.group(2)), "state", parent
)
continue
plain = re.match(r"^state\s+([\w.:-]+)$", text, re.I)
if plain:
diagram.add_node(plain.group(1), plain.group(1), "state", parent)
@@ -870,8 +881,13 @@ def analyze(diagram: Diagram) -> dict[str, Any]:
}
def _escape_markdown(text: str) -> str:
encoded = html.escape(text, quote=False)
return re.sub(r"([\\`*{}\[\]()#+\-.!_|>])", r"\\\1", encoded)
def _escape_table(text: str) -> str:
return text.replace("\n", " ⏎ ").replace("|", "\\|")
return _escape_markdown(text.replace("\n", " ⏎ "))
def digest(
@@ -913,28 +929,40 @@ def digest(
)
if diagram.fragments:
fragments = ", ".join(
f"{item['kind']}({item['label'] or 'unlabeled'})" for item in diagram.fragments
f"{item['kind']}({_escape_markdown(item['label'] or 'unlabeled')})"
for item in diagram.fragments
)
output.append(f"- fragments: {fragments}")
if diagram.notes:
output.append(f"- notes: {'; '.join(diagram.notes[:6])}")
output.append(
f"- notes: {'; '.join(_escape_markdown(note) for note in diagram.notes[:6])}"
)
if info["hubs"]:
output.append(
"- hubs (focal candidates): "
+ ", ".join(f"{hub['label']}({hub['degree']})" for hub in info["hubs"])
+ ", ".join(
f"{_escape_markdown(hub['label'])}({hub['degree']})"
for hub in info["hubs"]
)
)
if info["entry_points"]:
output.append(f"- entry points: {', '.join(info['entry_points'])}")
output.append(
f"- entry points: {', '.join(_escape_markdown(label) for label in info['entry_points'])}"
)
if info["terminals"]:
output.append(f"- terminals: {', '.join(info['terminals'])}")
output.append(
f"- terminals: {', '.join(_escape_markdown(label) for label in info['terminals'])}"
)
if info["orphans"]:
output.append(f"- unconnected: {', '.join(info['orphans'])}")
output.append(
f"- unconnected: {', '.join(_escape_markdown(label) for label in info['orphans'])}"
)
if info["collapsible_groups"]:
output.append("- collapsible groups (simplify here first):")
for group in info["collapsible_groups"]:
output.append(
f" - {group['label']} — {group['children']} children: "
+ ", ".join(group["child_labels"])
f" - {_escape_markdown(group['label'])} — {group['children']} children: "
+ ", ".join(_escape_markdown(label) for label in group["child_labels"])
)
output.extend(