diff --git a/graphify/extractors/sql.py b/graphify/extractors/sql.py
index d250956e0..840a52362 100644
--- a/graphify/extractors/sql.py
+++ b/graphify/extractors/sql.py
@@ -40,6 +40,25 @@ _ROUTINE_RECOVERY_RX = re.compile(
     re.IGNORECASE,
 )
 
+# A trigger with a procedural body (`... FOR EACH ROW BEGIN ... END`) has no
+# grammar production, so the whole statement lands in an ERROR node and the
+# trigger — plus the table it fires on — was dropped entirely. TRIGGER is
+# deliberately kept OUT of _ROUTINE_RECOVERY_RX (that path labels its matches
+# `name()`, but a trigger is not callable); it is recovered here instead, with
+# the subject table taken from the first ON after the name (the timing clause,
+# `BEFORE INSERT`, carries none). Same delimited-identifier atom as above.
+_TRIGGER_IDENT = (
+    r"(?:\"(?:[^\"\n]|\"\")+\"|\[(?:[^\]\n]|\]\])+\]|[\w$]+)"
+    r"(?:\s*\.\s*(?:\"(?:[^\"\n]|\"\")+\"|\[(?:[^\]\n]|\]\])+\]|[\w$]+))*"
+)
+_TRIGGER_RECOVERY_RX = re.compile(
+    r"\bCREATE\s+(?:OR\s+(?:REPLACE|ALTER)\s+)?TRIGGER\s+"
+    r"(?:IF\s+NOT\s+EXISTS\s+)?"
+    r"(" + _TRIGGER_IDENT + r")"
+    r".*?\bON\s+(" + _TRIGGER_IDENT + r")",
+    re.IGNORECASE | re.DOTALL,
+)
+
 # _mask_sql_comments is a linear character scanner, not a regex: the four
 # span kinds interact in ways a single pattern cannot express safely —
 # comment-opener parity inside strings, nested block comments, and
@@ -469,15 +488,18 @@ def extract_sql(path: Path, content: str | bytes | None = None) -> dict:
             trig_name: str | None = None
             tbl_name: str | None = None
             after_trigger = False
-            after_for = False
+            after_on = False
             for c in node.children:
                 if c.type == "keyword_trigger":
                     after_trigger = True
                 elif after_trigger and not trig_name and c.type == "object_reference":
                     trig_name = _read(c)
-                elif c.type == "keyword_for":
-                    after_for = True
-                elif after_for and not tbl_name and c.type == "object_reference":
+                # The subject table follows ON (`... AFTER INSERT ON users`), not
+                # FOR: `FOR EACH ROW` carries no table, so keying off keyword_for
+                # left every trigger without a link to the table it fires on.
+                elif c.type == "keyword_on":
+                    after_on = True
+                elif after_on and not tbl_name and c.type == "object_reference":
                     tbl_name = _read(c)
             if trig_name:
                 trig_nid = _make_id(stem, trig_name)
@@ -717,4 +739,19 @@ def extract_sql(path: Path, content: str | bytes | None = None) -> dict:
             fn_line = src_text[: m.start()].count("\n") + 1
             _add_node(_make_id(stem, fn_name), f"{fn_name}()", fn_line)
 
+        for m in _TRIGGER_RECOVERY_RX.finditer(masked_src):
+            if any(s <= m.start() < e for s, e in ident_spans):
+                continue
+            trig_name = m.group(1)
+            trig_nid = _make_id(stem, trig_name)
+            # A cleanly-parsed trigger elsewhere already owns this id (with its
+            # own ON edge); don't re-add and duplicate the edge.
+            if trig_nid in seen_ids:
+                continue
+            trig_line = src_text[: m.start()].count("\n") + 1
+            _add_node(trig_nid, trig_name, trig_line)
+            tbl_name = m.group(2)
+            tbl_nid = table_nids.get(_norm_ident(tbl_name)) or _ref_stub(tbl_name)
+            _add_edge(trig_nid, tbl_nid, "triggers", trig_line)
+
     return {"nodes": nodes, "edges": edges}
diff --git a/tests/test_multilang.py b/tests/test_multilang.py
index b9901f402..6d083e320 100644
--- a/tests/test_multilang.py
+++ b/tests/test_multilang.py
@@ -592,6 +592,57 @@ def test_sql_create_index_emits_index_node_linked_to_its_table(tmp_path):
     assert sum(1 for e in r["edges"] if e["relation"] == "indexes") == 4
 
 
+def test_sql_clean_trigger_links_to_its_on_table(tmp_path):
+    """A trigger's subject table follows ON, not FOR.
+
+    The create_trigger branch read the table off keyword_for, but `FOR EACH ROW`
+    carries no table — so a cleanly-parsed trigger got a node with no link to the
+    table it fires on.
+    """
+    pytest.importorskip("tree_sitter_sql")
+    p = tmp_path / "schema.sql"
+    p.write_text(
+        "CREATE TABLE users (id INT);\n"
+        "CREATE TRIGGER audit_ins AFTER INSERT ON users EXECUTE FUNCTION log_it();\n",
+        encoding="utf-8",
+    )
+    r = extract_sql(p)
+    by_label = {n["label"]: n for n in r["nodes"]}
+    assert "audit_ins" in by_label
+    edges = {(e["source"], e["relation"], e["target"]) for e in r["edges"]}
+    assert (by_label["audit_ins"]["id"], "triggers", by_label["users"]["id"]) in edges
+
+
+def test_sql_procedural_body_trigger_is_recovered(tmp_path):
+    """A trigger with a `FOR EACH ROW BEGIN ... END` body has no grammar parse,
+    so the statement lands in ERROR recovery. TRIGGER was excluded from the
+    routine-recovery pattern, so the whole trigger — and its table — was dropped.
+    """
+    pytest.importorskip("tree_sitter_sql")
+    p = tmp_path / "schema.sql"
+    p.write_text(
+        "CREATE TABLE users (id INT);\n"
+        "CREATE TABLE stats (cnt INT);\n"
+        "CREATE TRIGGER trg_after AFTER INSERT ON users\n"
+        "FOR EACH ROW\n"
+        "BEGIN\n"
+        "  UPDATE stats SET cnt = cnt + 1;\n"
+        "END;\n",
+        encoding="utf-8",
+    )
+    r = extract_sql(p)
+    by_label = {n["label"]: n for n in r["nodes"]}
+    assert "trg_after" in by_label, "procedural-body trigger dropped"
+    # a trigger is not callable — its label carries no ()
+    assert by_label["trg_after"]["label"] == "trg_after"
+    edges = {(e["source"], e["relation"], e["target"]) for e in r["edges"]}
+    assert (by_label["trg_after"]["id"], "triggers", by_label["users"]["id"]) in edges
+    # exactly one triggers edge, and nothing dangles
+    assert sum(1 for e in r["edges"] if e["relation"] == "triggers") == 1
+    node_ids = {n["id"] for n in r["nodes"]}
+    assert all(e["source"] in node_ids and e["target"] in node_ids for e in r["edges"])
+
+
 def test_sql_tsql_bracketed_procedure_is_recovered(tmp_path):
     """T-SQL CREATE PROCEDURE [Schema].[Name] ... AS BEGIN...END.