diff --git a/projekt-matching/projektmatch/ingest.py b/projekt-matching/projektmatch/ingest.py
new file mode 100644
index 0000000..961194d
--- /dev/null
+++ b/projekt-matching/projektmatch/ingest.py
@@ -0,0 +1,127 @@
+"""Flow 1: poll inbox, split projects, dedup, dispatch Flow 2, alert, trash."""
+from __future__ import annotations
+
+import email.header
+import fcntl
+import json
+import os
+
+import requests
+
+from . import espocrm, mailer, mailparse
+
+SUMMARY_KEYS = ("created", "rejected", "duplicate", "failed")
+LOCK_NAME = "projekt-matching.lock"
+
+
+def acquire_lock(data_dir):
+    path = os.path.join(data_dir, LOCK_NAME)
+    handle = open(path, "w")                      # noqa: SIM115
+    try:
+        fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        return handle
+    except OSError:
+        handle.close()
+        return None
+
+
+def release_lock(handle):
+    fcntl.flock(handle, fcntl.LOCK_UN)
+    handle.close()
+
+
+def decode_subject(msg):
+    raw = msg["Subject"] or ""
+    try:
+        parts = email.header.decode_header(raw)
+        return "".join(p.decode(enc or "utf-8", "replace")
+                       if isinstance(p, bytes) else p for p, enc in parts)
+    except Exception:
+        return raw
+
+
+def dispatch_flow2(cfg, item):
+    resp = requests.post(
+        f"{cfg.langflow_base}/api/v1/run/{cfg.flow2_id}?stream=false",
+        headers={"x-api-key": cfg.langflow_api_key},
+        json={"input_value": json.dumps(
+            {"canonical": item["canonical"], "title": item["title"]},
+            ensure_ascii=False),
+            "input_type": "text", "output_type": "text"},
+        timeout=1800)
+    resp.raise_for_status()
+    return parse_run_result(resp.json())
+
+
+def parse_run_result(payload):
+    """Depth-first search for the stage summary JSON in the run response."""
+    stack = [payload]
+    while stack:
+        node = stack.pop()
+        if isinstance(node, dict):
+            stack.extend(node.values())
+        elif isinstance(node, list):
+            stack.extend(node)
+        elif isinstance(node, str) and node.lstrip().startswith("{"):
+            try:
+                data = json.loads(node)
+            except json.JSONDecodeError:
+                continue
+            if isinstance(data, dict) and "status" in data:
+                return data
+    raise ValueError("Kein Status-JSON in der Run-Antwort gefunden")
+
+
+def run_ingest(cfg, mailbox=None, dispatch=None, espo=None, resolve=None):
+    lock = acquire_lock(cfg.data_dir)
+    if lock is None:
+        return {"skipped": "locked"}
+    try:
+        box = mailbox or mailer.MailBox(cfg.imap_user, cfg.imap_password)
+        crm = espo or espocrm.EspoClient(cfg.espo_base, cfg.espo_api_key)
+        send = dispatch or (lambda item: dispatch_flow2(cfg, item))
+        resolve = resolve or mailparse.canonical_url
+        summary = {"mails": 0, "projects": 0,
+                   **{k: 0 for k in SUMMARY_KEYS}}
+        for uid in box.unchecked_uids():
+            msg = box.fetch(uid)
+            subject = decode_subject(msg)
+            html, text = mailparse.bodies(msg)
+            items = [] if mailparse.is_own_mail(subject) else \
+                mailparse.split_projects(html, text)
+            if not items:
+                box.flag_checked(uid)
+                continue
+            summary["mails"] += 1
+            results = []
+            for item in items:
+                summary["projects"] += 1
+                result = dict(item)
+                try:
+                    result["canonical"] = resolve(item["url"])
+                    dup = crm.find_opportunity_by_link(result["canonical"])
+                    if dup:
+                        result["status"] = "duplicate"
+                    else:
+                        result.update(send(result))
+                except Exception as exc:          # noqa: BLE001
+                    result.update(status="failed", error=str(exc))
+                results.append(result)
+                summary[result["status"]] += 1
+            failed = [r for r in results if r["status"] == "failed"]
+            if failed:
+                lines = "\n\n".join(
+                    f"- {f.get('title', '?')}\n"
+                    f"  {f.get('canonical', f.get('url', '?'))}\n"
+                    f"  Fehler: {f.get('error', '?')}" for f in failed)
+                mailer.send_mail(
+                    cfg.imap_user, cfg.imap_password, cfg.alert_to,
+                    f"[Projekt-Match-Fehler] Freelancermap — "
+                    f"{len(failed)} Projekt(e)",
+                    "Folgende Projekte konnten nicht verarbeitet "
+                    "werden:\n\n" + lines + "\n")
+            box.move_to_trash(uid)
+        box.close()
+        return summary
+    finally:
+        release_lock(lock)
diff --git a/projekt-matching/tests/test_ingest.py b/projekt-matching/tests/test_ingest.py
new file mode 100644
index 0000000..94c14ef
--- /dev/null
+++ b/projekt-matching/tests/test_ingest.py
@@ -0,0 +1,97 @@
+import json
+from unittest import mock
+
+from projektmatch import ingest
+from projektmatch.config import Cfg
+
+
+PROJECT_HTML = ('<a href="https://www.freelancermap.de/nproj/1.html?x=1">'
+                'Projekt Eins</a>')
+
+
+def cfg(tmp_path):
+    return Cfg(imap_password="pw", alert_to="a@x", flow2_id="F2",
+               langflow_api_key="key", data_dir=str(tmp_path))
+
+
+def run(tmp_path, mails, dup=None, dispatch_result=None):
+    box = mock.Mock()
+    box.unchecked_uids.return_value = list(mails)
+    box.fetch.side_effect = lambda uid: mails[uid]
+    espo = mock.Mock()
+    espo.find_opportunity_by_link.return_value = dup
+    sent = []
+    with mock.patch("projektmatch.ingest.mailparse.bodies",
+                    side_effect=lambda m: (m["html"], "")), \
+         mock.patch("projektmatch.ingest.mailer.send_mail",
+                    lambda *a: sent.append(a)):
+        summary = ingest.run_ingest(
+            cfg(tmp_path), mailbox=box, espo=espo,
+            resolve=lambda url: "https://www.freelancermap.de/projekt/eins",
+            dispatch=lambda item: dispatch_result or {"status": "created"})
+    return summary, box, sent
+
+
+def mail(subject, html=""):
+    return {"Subject": subject, "html": html}
+
+
+def test_non_project_mail_flagged_not_trashed(tmp_path):
+    mails = {"1": mail("Newsletter", "<p>Hallo</p>")}
+    summary, box, sent = run(tmp_path, mails)
+    box.flag_checked.assert_called_once_with("1")
+    box.move_to_trash.assert_not_called()
+    assert summary["mails"] == 0 and not sent
+
+
+def test_own_mail_skipped(tmp_path):
+    mails = {"1": mail("[Projekt-Match] Freelancermap — 90 % — X",
+                       PROJECT_HTML)}
+    summary, box, sent = run(tmp_path, mails)
+    box.flag_checked.assert_called_once_with("1")
+    box.move_to_trash.assert_not_called()
+
+
+def test_project_mail_created_and_trashed(tmp_path):
+    mails = {"1": mail("Example 1 for Langflow process", PROJECT_HTML)}
+    summary, box, sent = run(tmp_path, mails)
+    assert summary == {"mails": 1, "projects": 1, "created": 1,
+                       "rejected": 0, "duplicate": 0, "failed": 0}
+    box.move_to_trash.assert_called_once_with("1")
+    assert not sent
+
+
+def test_duplicate_skips_dispatch(tmp_path):
+    mails = {"1": mail("Mail", PROJECT_HTML)}
+    summary, box, sent = run(tmp_path, mails, dup={"id": "X"})
+    assert summary["duplicate"] == 1 and summary["created"] == 0
+    box.move_to_trash.assert_called_once()
+
+
+def test_failed_dispatch_alerts_and_trashes(tmp_path):
+    mails = {"1": mail("Mail", PROJECT_HTML)}
+    summary, box, sent = run(tmp_path, mails,
+                             dispatch_result={"status": "failed",
+                                              "error": "vLLM down"})
+    assert summary["failed"] == 1
+    assert len(sent) == 1
+    _, _, to, subject, body = sent[0]
+    assert to == "a@x"
+    assert subject == "[Projekt-Match-Fehler] Freelancermap — 1 Projekt(e)"
+    assert "vLLM down" in body
+    box.move_to_trash.assert_called_once()
+
+
+def test_lock_prevents_second_run(tmp_path):
+    lock = ingest.acquire_lock(str(tmp_path))
+    assert lock is not None
+    summary = ingest.run_ingest(cfg(tmp_path), mailbox=mock.Mock())
+    assert summary == {"skipped": "locked"}
+    ingest.release_lock(lock)
+
+
+def test_parse_run_result_finds_status_json():
+    inner = json.dumps({"status": "created", "mustMatch": 90})
+    payload = {"outputs": [{"outputs": [{"results": {"text": {
+        "data": {"text": inner}}}}]}]}
+    assert ingest.parse_run_result(payload)["status"] == "created"