diff --git a/projekt-matching/projektmatch/pm_trace.py b/projekt-matching/projektmatch/pm_trace.py
new file mode 100644
index 0000000..6b7601c
--- /dev/null
+++ b/projekt-matching/projektmatch/pm_trace.py
@@ -0,0 +1,43 @@
+"""Post one consolidated Langfuse trace per project (ingestion API).
+
+Reads the LANGFUSE_* env vars that create_pod_langflow.sh already sets on
+the Langflow container. Never raises — tracing must not fail a run."""
+from __future__ import annotations
+
+import datetime as dt
+import os
+import uuid
+
+import requests
+
+
+def post_trace(ctx: dict) -> None:
+    host = os.environ.get("LANGFUSE_HOST")
+    pk = os.environ.get("LANGFUSE_PUBLIC_KEY")
+    sk = os.environ.get("LANGFUSE_SECRET_KEY")
+    if not (host and pk and sk):
+        return
+    now = dt.datetime.now(dt.timezone.utc).isoformat()
+    body = {
+        "id": str(uuid.uuid4()),
+        "timestamp": now,
+        "name": "projekt-match",
+        "input": {"canonical": ctx.get("canonical"), "title": ctx.get("title")},
+        "output": {"decision": ctx.get("decision"),
+                   "status": ctx.get("status"),
+                   "mustMatch": ctx.get("mustMatch"),
+                   "niceMatch": ctx.get("niceMatch"),
+                   "description": ctx.get("description")},
+        "metadata": {"opportunityId": ctx.get("opportunityId"),
+                     "crmUrl": ctx.get("crmUrl"),
+                     "error": ctx.get("error"),
+                     "projectName": ctx.get("projectName")},
+        "tags": [ctx.get("status") or "unknown"],
+    }
+    event = {"batch": [{"id": str(uuid.uuid4()), "type": "trace-create",
+                        "timestamp": now, "body": body}]}
+    try:
+        requests.post(f"{host.rstrip('/')}/api/public/ingestion", json=event,
+                      auth=(pk, sk), timeout=15)
+    except Exception:
+        pass
diff --git a/projekt-matching/projektmatch/stages.py b/projekt-matching/projektmatch/stages.py
new file mode 100644
index 0000000..46389a7
--- /dev/null
+++ b/projekt-matching/projektmatch/stages.py
@@ -0,0 +1,166 @@
+"""Flow 2 stage functions. Each takes (ctx, cfg) and returns ctx.
+
+ctx["status"]: "ok" -> pipeline continues; "failed" -> all later stages
+skip; "rejected" -> only CRM/notify skip; "created" -> set by stage_crm.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import json
+import time
+from pathlib import Path
+
+import requests
+from bs4 import BeautifulSoup
+
+from . import espocrm, llm, mailer, pm_trace, rules
+from .mailparse import BROWSER_UA
+
+TEAM_BY_OFFER = {"Projekt": "DesTEngS",
+                 "Arbeitnehmer-Angebot": "Arbeitnehmer",
+                 "ANÜ": "ANÜ"}
+MIN_PAGE_CHARS = 200
+
+
+def run_stage(name, fn, ctx, cfg, **kw):
+    if ctx.get("status") == "failed":
+        return ctx
+    try:
+        return fn(ctx, cfg, **kw)
+    except Exception as exc:                     # noqa: BLE001
+        ctx.update(status="failed", error=f"{name}: {exc}")
+        return ctx
+
+
+def stage_fetch(ctx, cfg):
+    last = None
+    for attempt in range(3):
+        try:
+            resp = requests.get(ctx["canonical"], timeout=30,
+                                headers={"User-Agent": BROWSER_UA})
+            resp.raise_for_status()
+            soup = BeautifulSoup(resp.text, "html.parser")
+            for tag in soup(["script", "style", "noscript"]):
+                tag.decompose()
+            text = "\n".join(line for line in
+                             soup.get_text("\n", strip=True).splitlines()
+                             if line.strip())
+            if len(text) < MIN_PAGE_CHARS:
+                raise ValueError(
+                    f"Seitentext zu kurz ({len(text)} Zeichen) — "
+                    f"Login-Wall oder leere Seite?")
+            ctx.update(page_text=text[:llm.PAGE_TEXT_LIMIT], status="ok")
+            return ctx
+        except Exception as exc:                 # noqa: BLE001
+            last = exc
+            time.sleep(2 * (attempt + 1))
+    raise RuntimeError(f"Abruf fehlgeschlagen: {last}")
+
+
+def stage_extract(ctx, cfg):
+    ctx["extract"] = llm.extract_project(cfg.vllm_base, cfg.vllm_model,
+                                         ctx["page_text"])
+    ctx["projectName"] = ctx["extract"]["projectName"]
+    return ctx
+
+
+def stage_match(ctx, cfg):
+    reqs = [{"nr": i + 1, "text": r["text"]}
+            for i, r in enumerate(ctx["extract"]["requirements"])
+            if r["kat"] in ("Must", "Nice")]
+    if not reqs:
+        ctx["ratings"] = []
+        return ctx
+    cv_path = Path(cfg.data_dir) / "vorgaben" / \
+        "Lebenslauf_Dr-Ing_Thomas_Langer.md"
+    cv_text = cv_path.read_text(encoding="utf-8")
+    ctx["ratings"] = llm.match_cv(cfg.vllm_base, cfg.vllm_model, cv_text, reqs)
+    return ctx
+
+
+def stage_rules(ctx, cfg):
+    today = dt.date.today()
+    rated = {r["nr"]: r["rating"] for r in ctx.get("ratings", [])}
+    rows, nr = [], 0
+    for req in ctx["extract"]["requirements"]:
+        if req["kat"] in ("Must", "Nice"):
+            nr += 1
+            rating = rated.get(nr, rules.UNKNOWN)
+        else:
+            rating = rules.eval_misc(req, today)
+        rows.append({"kat": req["kat"], "rating": rating, "text": req["text"]})
+    rows = rules.order_rows(rows)
+    ctx["mustMatch"] = rules.calc_match(rows, "Must")
+    ctx["niceMatch"] = rules.calc_match(rows, "Nice")
+    ctx["description"] = rules.build_description(rows)
+    consider = ctx["mustMatch"] is not None and \
+        ctx["mustMatch"] > int(cfg.threshold)
+    ctx["decision"] = "consider" if consider else "rejected"
+    if not consider:
+        ctx["status"] = "rejected"
+    return ctx
+
+
+def stage_crm(ctx, cfg, espo=None):
+    if ctx.get("status") != "ok" or ctx.get("decision") != "consider":
+        return ctx
+    espo = espo or espocrm.EspoClient(cfg.espo_base, cfg.espo_api_key)
+    ex = ctx["extract"]
+    team = espo.team_id(TEAM_BY_OFFER[ex["offerType"]])
+    agency = ex["buyerType"] == "agency"
+    account_id = None
+    if ex.get("companyName"):
+        account_id = espo.ensure_account(
+            ex["companyName"], "Reseller" if agency else "Customer")
+    contact_id = None
+    if ex.get("contactPerson") and account_id:
+        first, last = espocrm.split_person(ex["contactPerson"])
+        contact_id = espo.ensure_contact(first, last, account_id)
+    name = espo.unique_opportunity_name(ex["projectName"])
+    payload = {"name": name, "description": ctx["description"],
+               "cProjektlink": ctx["canonical"], "teamsIds": [team]}
+    if account_id:
+        payload["cAccount1Id" if agency else "accountId"] = account_id
+    if contact_id:
+        payload["contactsIds"] = [contact_id]
+    created = espo.create_opportunity(payload)
+    oid = created["id"]
+    back = espo.get_opportunity(oid)
+    problems = []
+    if back.get("name") != name:
+        problems.append("name")
+    if back.get("cProjektlink") != ctx["canonical"]:
+        problems.append("cProjektlink")
+    if back.get("description") != ctx["description"]:
+        problems.append("description")
+    if account_id and agency and not back.get("cAccount1Id"):
+        problems.append("cAccount1Id")
+    if account_id and not agency and not back.get("accountId"):
+        problems.append("accountId")
+    if problems:
+        raise RuntimeError(f"CRM-Verifikation fehlgeschlagen: {problems}")
+    ctx.update(status="created", opportunityId=oid, projectName=name,
+               crmUrl=f"{cfg.crm_web_base}/#Opportunity/view/{oid}")
+    return ctx
+
+
+def stage_notify(ctx, cfg):
+    if ctx.get("status") == "created":
+        subject = (f"[Projekt-Match] Freelancermap — {ctx['mustMatch']} % — "
+                   f"{ctx['projectName']}")
+        body = (f"Neues passendes Projekt gefunden.\n\n"
+                f"Projekt: {ctx['projectName']}\n"
+                f"Must-have-Match: {rules.fmt_match(ctx['mustMatch'])}\n"
+                f"Nice-to-have-Match: {rules.fmt_match(ctx.get('niceMatch'))}\n\n"
+                f"CRM-Verkaufschance: {ctx['crmUrl']}\n")
+        mailer.send_mail(cfg.imap_user, cfg.imap_password, cfg.notify_to,
+                         subject, body)
+    pm_trace.post_trace(ctx)
+    return ctx
+
+
+def summary(ctx):
+    """Compact JSON result string for TextOutput / Flow 1."""
+    keys = ("status", "error", "decision", "mustMatch", "niceMatch",
+            "opportunityId", "crmUrl", "projectName", "canonical")
+    return json.dumps({k: ctx.get(k) for k in keys}, ensure_ascii=False)
diff --git a/projekt-matching/tests/test_stages.py b/projekt-matching/tests/test_stages.py
new file mode 100644
index 0000000..91767e2
--- /dev/null
+++ b/projekt-matching/tests/test_stages.py
@@ -0,0 +1,102 @@
+import datetime as dt
+from unittest import mock
+
+from projektmatch import stages
+from projektmatch.config import Cfg
+
+CFG = Cfg(imap_password="pw", espo_base="https://crm/api/v1",
+          espo_api_key="k", notify_to="n@x", alert_to="a@x",
+          vllm_base="http://v/v1", vllm_model="m", data_dir="/tmp")
+
+EXTRACT = {
+    "projectName": "Python Entwickler KI",
+    "offerType": "Projekt", "buyerType": "agency",
+    "companyName": "Aristo Group", "contactPerson": "Max Mann",
+    "requirements": [
+        {"text": "Python", "kat": "Must", "miscType": "other",
+         "startDate": None, "workloadPercent": None, "remotePercent": None,
+         "onsiteLocation": None},
+        {"text": "K8s", "kat": "Nice", "miscType": "other",
+         "startDate": None, "workloadPercent": None, "remotePercent": None,
+         "onsiteLocation": None},
+        {"text": "100 % remote", "kat": "Misc", "miscType": "location",
+         "startDate": None, "workloadPercent": None, "remotePercent": 100,
+         "onsiteLocation": None},
+    ],
+}
+
+
+def ctx_after_match(ratings):
+    ctx = {"canonical": "https://x/projekt/p", "title": "t", "status": "ok",
+           "extract": dict(EXTRACT)}
+    ctx["ratings"] = ratings
+    return ctx
+
+
+def test_run_stage_catches_and_skips():
+    def boom(ctx, cfg):
+        raise ValueError("kaputt")
+    ctx = stages.run_stage("fetch", boom, {"status": "ok"}, CFG)
+    assert ctx["status"] == "failed" and "fetch: kaputt" in ctx["error"]
+    untouched = stages.run_stage("extract", boom, dict(ctx), CFG)
+    assert untouched["status"] == "failed"      # boom not called again
+
+
+def test_stage_rules_consider_and_reject():
+    yes = [{"nr": 1, "rating": "yes", "reason": "ok"},
+           {"nr": 2, "rating": "yes", "reason": "ok"}]
+    ctx = stages.stage_rules(ctx_after_match(yes), CFG)
+    assert ctx["decision"] == "consider" and ctx["mustMatch"] == 100
+    assert ctx["status"] == "ok"
+    assert "| 1 | Must | ✅ | Python |" in ctx["description"]
+    no = [{"nr": 1, "rating": "no", "reason": "fehlt"},
+          {"nr": 2, "rating": "yes", "reason": "ok"}]
+    ctx2 = stages.stage_rules(ctx_after_match(no), CFG)
+    assert ctx2["status"] == "rejected" and ctx2["decision"] == "rejected"
+    assert ctx2["mustMatch"] == 0
+
+
+def test_stage_crm_agency_linking_and_verify():
+    ctx = ctx_after_match([{"nr": 1, "rating": "yes", "reason": "ok"},
+                           {"nr": 2, "rating": "yes", "reason": "ok"}])
+    ctx = stages.stage_rules(ctx, CFG)
+    espo = mock.Mock()
+    espo.team_id.return_value = "T1"
+    espo.ensure_account.return_value = "A1"
+    espo.ensure_contact.return_value = "C1"
+    espo.unique_opportunity_name.return_value = "Python Entwickler KI"
+    espo.create_opportunity.return_value = {"id": "O1"}
+    espo.get_opportunity.return_value = {
+        "id": "O1", "name": "Python Entwickler KI",
+        "cProjektlink": "https://x/projekt/p",
+        "description": ctx["description"], "cAccount1Id": "A1",
+        "accountId": None, "teamsIds": ["T1"]}
+    out = stages.stage_crm(ctx, CFG, espo=espo)
+    assert out["status"] == "created" and out["opportunityId"] == "O1"
+    assert out["crmUrl"].endswith("#Opportunity/view/O1")
+    payload = espo.create_opportunity.call_args.args[0]
+    assert payload["cAccount1Id"] == "A1" and "accountId" not in payload
+    assert payload["teamsIds"] == ["T1"]
+    espo.ensure_account.assert_called_with("Aristo Group", "Reseller")
+
+
+def test_stage_crm_skips_when_rejected():
+    ctx = {"status": "rejected", "decision": "rejected"}
+    espo = mock.Mock()
+    assert stages.stage_crm(ctx, CFG, espo=espo)["status"] == "rejected"
+    espo.team_id.assert_not_called()
+
+
+def test_stage_notify_only_on_created():
+    sent = []
+    with mock.patch("projektmatch.stages.mailer.send_mail",
+                    lambda *a: sent.append(a)), \
+         mock.patch("projektmatch.stages.pm_trace.post_trace", lambda c: None):
+        ctx = {"status": "created", "projectName": "P", "mustMatch": 90,
+               "niceMatch": None, "crmUrl": "https://crm/#Opportunity/view/O1"}
+        stages.stage_notify(dict(ctx), CFG)
+        assert sent[0][2] == "n@x"
+        assert sent[0][3] == "[Projekt-Match] Freelancermap — 90 % — P"
+        sent.clear()
+        stages.stage_notify({"status": "rejected"}, CFG)
+        assert not sent