Merge branch 'projekt-matching': automated Langflow project-skill matching with CRM entry

All 16 plan tasks complete; acceptance gates 1-3 passed live; final review fixes applied.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

create_pod_langflow.sh

@@ -64,6 +64,12 @@ LANGFUSE_INIT_PUBLIC_KEY="pk-lf-00000000-0000-0000-0000-000000000001"
 LANGFUSE_INIT_SECRET_KEY="sk-lf-00000000-0000-0000-0000-000000000002"
 LANGFUSE_INIT_EMAIL="admin@example.com"
 LANGFUSE_INIT_PASSWORD="langfuse-admin-pw"
+# Admin key for the Organization Management API (used once by
+# projekt-matching/deploy/setup_langfuse.py to create the projekt-matching
+# project); PM_* keys are the Langfuse project keys Langflow traces to.
+LANGFUSE_ADMIN_API_KEY="admin-pm-3f61c2a89d4e"
+PM_LANGFUSE_PUBLIC_KEY="pk-lf-e12a9a7b-a87d-420c-b047-fd00885695eb"
+PM_LANGFUSE_SECRET_KEY="sk-lf-d0d55349-79fd-446d-b589-3ac4f35a0064"
 # Langflow sends its run traces to the in-pod Langfuse using the keys above.
 # NOTE: the Langfuse web (Next.js) server binds to the pod's hostname/IP, NOT
 # 127.0.0.1, so Langflow must address it via the pod name (which every pod
@@ -243,6 +249,16 @@ echo "Container '$REDIS_CTR_NAME' started (rc=$?)"
 # Podman automatically provides the `host.containers.internal` hostname in the
 # container's /etc/hosts (pointing at the host), so flows can reach the local
 # vLLM OpenAI-compatible endpoint on the host without an explicit --add-host.
+# PYTHONPATH=/app/langflow: durable equivalent of the projektmatch.pth file
+# deploy_files.sh writes into the venv's ephemeral site-packages. This unit is
+# generated with `podman generate systemd --new` (--rm + --replace in
+# ExecStart), so ANY restart of this container -- including the `podman
+# restart` deploy_files.sh issues after writing the .pth -- makes systemd
+# recreate it from the image, wiping the just-written .pth before the live
+# Langflow process (or any later exec) can ever observe it (verified: two
+# consecutive deploy_files.sh runs never converged). Setting PYTHONPATH here
+# survives every recreation because it's baked into the podman run command
+# itself, so "import projektmatch" works from the very first boot.
 podman run -d --name "$LANGFLOW_CTR_NAME" --pod "$POD_NAME" \
   -e LANGFLOW_DATABASE_URL="$LANGFLOW_DB_URL" \
   -e LANGFLOW_CONFIG_DIR=/app/langflow \
@@ -251,8 +267,9 @@ podman run -d --name "$LANGFLOW_CTR_NAME" --pod "$POD_NAME" \
   -e LANGFLOW_PRETTY_LOGS=true \
   -e LANGFLOW_LOG_LEVEL=INFO \
   -e DO_NOT_TRACK=True \
-  -e LANGFUSE_PUBLIC_KEY="$LANGFUSE_INIT_PUBLIC_KEY" \
-  -e LANGFUSE_SECRET_KEY="$LANGFUSE_INIT_SECRET_KEY" \
+  -e PYTHONPATH=/app/langflow \
+  -e LANGFUSE_PUBLIC_KEY="$PM_LANGFUSE_PUBLIC_KEY" \
+  -e LANGFUSE_SECRET_KEY="$PM_LANGFUSE_SECRET_KEY" \
   -e LANGFUSE_HOST="$LANGFUSE_INPOD_HOST" \
   -v "$LANGFLOW_DATA_DIR:/app/langflow:Z" \
   "$LANGFLOW_IMAGE"
@@ -267,6 +284,7 @@ echo "Container '$LANGFUSE_WORKER_CTR_NAME' started (rc=$?)"
 # Langfuse web container (exposed on host port 8091 -> 3000)
 podman run -d --name "$LANGFUSE_WEB_CTR_NAME" --pod "$POD_NAME" \
   "${LANGFUSE_COMMON_ENV[@]}" \
+  -e ADMIN_API_KEY="$LANGFUSE_ADMIN_API_KEY" \
   -e NEXTAUTH_SECRET="qwexczutbewrgerznupvemqyw" \
   -e LANGFUSE_INIT_ORG_ID="$LANGFUSE_INIT_ORG" \
   -e LANGFUSE_INIT_ORG_NAME="Test Org" \

docs/superpowers/plans/2026-07-09-langflow-projekt-matching.md

@@ -25,6 +25,18 @@
 - Langflow API auth (v1.10.0): flow CRUD + variables need Bearer JWT from `GET /api/v1/auto_login`; `POST /api/v1/run/...` and webhook need `x-api-key`. Responses are gzip → `curl --compressed`.
 - Langflow data dir: host `~/.local/share/langflow_pod/langflow-data` = container `/app/langflow`. Deployed files need group-read (dir is `g+rwxs`, uid-1000/gid-0 process).
 
+## Deviations discovered during execution
+
+- `response_format json_schema` replaces `guided_json` (vLLM 0.22 ignores `guided_json`).
+- `chat_template_kwargs` `enable_thinking: false` is required for structured calls.
+- Component input `ctx` was renamed `ctx_in` (Langflow reserves `Component.ctx`).
+- `PYTHONPATH=/app/langflow` env var replaces the `.pth`-file approach.
+- Langfuse org/project provisioning goes through the session tRPC endpoint (the admin API is EE-gated).
+- Flow 1 is a single `PMIngest` component (not the originally sketched multi-node graph).
+- `PMNotify` calls `stage_notify` directly rather than going through an intermediate layer.
+
+Task-level code snippets further below in this document were NOT retro-edited to reflect these deviations; the `projektmatch` package source under `/home/lwc/bin/projekt-matching/` is authoritative.
+
 ## File Structure
 
 ```
@@ -1901,7 +1913,7 @@ cd /home/lwc/bin && git add projekt-matching && git commit -m "feat(projekt-matc
 
 **Interfaces:**
 - Consumes: `projektmatch.stages`, `projektmatch.ingest`, `projektmatch.config.Cfg` (deployed at `/app/langflow/projektmatch` inside the container, Task 10).
-- Produces: seven Langflow `Component` classes. Class names / output names are consumed by `build_flows.py` (Task 10): every component has exactly one output named `out`, method `build_out`. Global-variable names referenced via `load_from_db`: `PM_IMAP_PASSWORD`, `PM_ESPO_API_KEY` (Credential); `PM_ESPO_BASE`, `PM_NOTIFY_TO`, `PM_ALERT_TO`, `PM_THRESHOLD`, `PM_VLLM_BASE`, `PM_VLLM_MODEL`, `PM_FLOW2_ID`, `PM_LANGFLOW_API_KEY` (Generic).
+- Produces: seven Langflow `Component` classes. Class names / output names are consumed by `build_flows.py` (Task 10): every component has exactly one output named `out`, method `build_out`. Global-variable names referenced via `load_from_db`: `PM_IMAP_PASSWORD`, `PM_ESPO_API_KEY`, `PM_LANGFLOW_API_KEY` (Credential, SecretStrInput); `PM_ESPO_BASE`, `PM_NOTIFY_TO`, `PM_ALERT_TO`, `PM_THRESHOLD`, `PM_VLLM_BASE`, `PM_VLLM_MODEL`, `PM_FLOW2_ID` (Generic, MessageTextInput).
 - No host-side unit tests (wrappers only; logic is tested in Tasks 2–8). Verified live in Task 10 Step 6 and Task 13.
 
 - [ ] **Step 1: Write `components/pm_fetch.py`**
@@ -2119,8 +2131,9 @@ class PMNotify(Component):
 
     def build_out(self) -> Message:
         cfg = Cfg(imap_password=self.imap_password, notify_to=self.notify_to)
-        ctx = stages.run_stage("notify", stages.stage_notify,
-                               dict(self.ctx.data), cfg)
+        # stage_notify is called DIRECTLY (not via run_stage): it must run for
+        # every terminal status so failed runs still post their Langfuse trace.
+        ctx = stages.stage_notify(dict(self.ctx.data), cfg)
         self.status = ctx.get("status", "")
         return Message(text=stages.summary(ctx))
 ```
@@ -2374,6 +2387,9 @@ def main():
         if name != "PM_FLOW2_ID":
             upsert_variable(hdr, name, value, "Generic")
     api_key = get_api_key(hdr)
+    # PMIngest reads the run-API key via load_from_db -> must exist as a
+    # Credential global variable, not only in the host env file.
+    upsert_variable(hdr, "PM_LANGFLOW_API_KEY", api_key, "Credential")
 
     def custom(fname):
         code = (COMPONENTS / fname).read_text()

projekt-matching/.gitignore

@@ -0,0 +1,5 @@
+.venv/
+__pycache__/
+*.pyc
+*.local.env
+*.egg-info/

projekt-matching/components/pm_crm.py

@@ -0,0 +1,33 @@
+import sys
+
+if "/app/langflow" not in sys.path:
+    sys.path.insert(0, "/app/langflow")
+
+from langflow.custom import Component
+from langflow.io import DataInput, MessageTextInput, Output, SecretStrInput
+from langflow.schema import Data
+
+from projektmatch import stages
+from projektmatch.config import Cfg
+
+
+class PMCrm(Component):
+    display_name = "PM 5 CRM"
+    description = "EspoCRM: Team, Firma, Kontakt, Verkaufschance + Verify"
+    inputs = [
+        DataInput(name="ctx_in", display_name="Context"),
+        MessageTextInput(name="espo_base", display_name="Espo Base",
+                         value="PM_ESPO_BASE", load_from_db=True,
+                         advanced=True),
+        SecretStrInput(name="espo_api_key", display_name="Espo API Key",
+                       value="PM_ESPO_API_KEY", load_from_db=True,
+                       advanced=True),
+    ]
+    outputs = [Output(name="out", display_name="Context", method="build_out")]
+
+    def build_out(self) -> Data:
+        cfg = Cfg(espo_base=self.espo_base, espo_api_key=self.espo_api_key)
+        ctx = stages.run_stage("crm", stages.stage_crm,
+                               dict(self.ctx_in.data), cfg)
+        self.status = ctx.get("status", "")
+        return Data(data=ctx)

projekt-matching/components/pm_extract.py

@@ -0,0 +1,33 @@
+import sys
+
+if "/app/langflow" not in sys.path:
+    sys.path.insert(0, "/app/langflow")
+
+from langflow.custom import Component
+from langflow.io import DataInput, MessageTextInput, Output
+from langflow.schema import Data
+
+from projektmatch import stages
+from projektmatch.config import Cfg
+
+
+class PMExtract(Component):
+    display_name = "PM 2 Extract"
+    description = "LLM 1: Anforderungen strukturiert extrahieren"
+    inputs = [
+        DataInput(name="ctx_in", display_name="Context"),
+        MessageTextInput(name="vllm_base", display_name="vLLM Base",
+                         value="PM_VLLM_BASE", load_from_db=True,
+                         advanced=True),
+        MessageTextInput(name="vllm_model", display_name="vLLM Model",
+                         value="PM_VLLM_MODEL", load_from_db=True,
+                         advanced=True),
+    ]
+    outputs = [Output(name="out", display_name="Context", method="build_out")]
+
+    def build_out(self) -> Data:
+        cfg = Cfg(vllm_base=self.vllm_base, vllm_model=self.vllm_model)
+        ctx = stages.run_stage("extract", stages.stage_extract,
+                               dict(self.ctx_in.data), cfg)
+        self.status = ctx.get("status", "")
+        return Data(data=ctx)

projekt-matching/components/pm_fetch.py

@@ -0,0 +1,26 @@
+import sys
+
+if "/app/langflow" not in sys.path:
+    sys.path.insert(0, "/app/langflow")
+
+import json
+
+from langflow.custom import Component
+from langflow.io import MessageTextInput, Output
+from langflow.schema import Data
+
+from projektmatch import stages
+
+
+class PMFetch(Component):
+    display_name = "PM 1 Fetch"
+    description = "Projektseite abrufen (canonical URL -> Seitentext)"
+    inputs = [MessageTextInput(name="payload", display_name="Payload JSON")]
+    outputs = [Output(name="out", display_name="Context", method="build_out")]
+
+    def build_out(self) -> Data:
+        ctx = json.loads(self.payload)
+        ctx.setdefault("status", "ok")
+        ctx = stages.run_stage("fetch", stages.stage_fetch, ctx, None)
+        self.status = ctx.get("status", "")
+        return Data(data=ctx)

projekt-matching/components/pm_ingest.py

@@ -0,0 +1,52 @@
+import sys
+
+if "/app/langflow" not in sys.path:
+    sys.path.insert(0, "/app/langflow")
+
+import json
+
+from langflow.custom import Component
+from langflow.io import DataInput, MessageTextInput, Output, SecretStrInput
+from langflow.schema.message import Message
+
+from projektmatch import ingest
+from projektmatch.config import Cfg
+
+
+class PMIngest(Component):
+    display_name = "PM Ingest"
+    description = ("IMAP-Postfach abrufen, Projekte splitten, CRM-Dedup, "
+                   "Flow 2 je Projekt, Alert-Mail, Trigger-Mail -> Trash")
+    inputs = [
+        DataInput(name="trigger", display_name="Webhook Trigger"),
+        SecretStrInput(name="imap_password", display_name="Mail Password",
+                       value="PM_IMAP_PASSWORD", load_from_db=True,
+                       advanced=True),
+        MessageTextInput(name="espo_base", display_name="Espo Base",
+                         value="PM_ESPO_BASE", load_from_db=True,
+                         advanced=True),
+        SecretStrInput(name="espo_api_key", display_name="Espo API Key",
+                       value="PM_ESPO_API_KEY", load_from_db=True,
+                       advanced=True),
+        MessageTextInput(name="alert_to", display_name="Alert To",
+                         value="PM_ALERT_TO", load_from_db=True,
+                         advanced=True),
+        MessageTextInput(name="flow2_id", display_name="Flow 2 ID",
+                         value="PM_FLOW2_ID", load_from_db=True,
+                         advanced=True),
+        SecretStrInput(name="langflow_api_key", display_name="Langflow Key",
+                       value="PM_LANGFLOW_API_KEY", load_from_db=True,
+                       advanced=True),
+    ]
+    outputs = [Output(name="out", display_name="Summary", method="build_out")]
+
+    def build_out(self) -> Message:
+        cfg = Cfg(imap_password=self.imap_password,
+                  espo_base=self.espo_base,
+                  espo_api_key=self.espo_api_key,
+                  alert_to=self.alert_to,
+                  flow2_id=self.flow2_id,
+                  langflow_api_key=self.langflow_api_key)
+        summary = ingest.run_ingest(cfg)
+        self.status = json.dumps(summary)
+        return Message(text=json.dumps(summary))

projekt-matching/components/pm_match.py

@@ -0,0 +1,33 @@
+import sys
+
+if "/app/langflow" not in sys.path:
+    sys.path.insert(0, "/app/langflow")
+
+from langflow.custom import Component
+from langflow.io import DataInput, MessageTextInput, Output
+from langflow.schema import Data
+
+from projektmatch import stages
+from projektmatch.config import Cfg
+
+
+class PMMatch(Component):
+    display_name = "PM 3 Match CV"
+    description = "LLM 2: Anforderungen gegen Lebenslauf bewerten"
+    inputs = [
+        DataInput(name="ctx_in", display_name="Context"),
+        MessageTextInput(name="vllm_base", display_name="vLLM Base",
+                         value="PM_VLLM_BASE", load_from_db=True,
+                         advanced=True),
+        MessageTextInput(name="vllm_model", display_name="vLLM Model",
+                         value="PM_VLLM_MODEL", load_from_db=True,
+                         advanced=True),
+    ]
+    outputs = [Output(name="out", display_name="Context", method="build_out")]
+
+    def build_out(self) -> Data:
+        cfg = Cfg(vllm_base=self.vllm_base, vllm_model=self.vllm_model)
+        ctx = stages.run_stage("match", stages.stage_match,
+                               dict(self.ctx_in.data), cfg)
+        self.status = ctx.get("status", "")
+        return Data(data=ctx)

projekt-matching/components/pm_notify.py

@@ -0,0 +1,34 @@
+import sys
+
+if "/app/langflow" not in sys.path:
+    sys.path.insert(0, "/app/langflow")
+
+from langflow.custom import Component
+from langflow.io import DataInput, MessageTextInput, Output, SecretStrInput
+from langflow.schema.message import Message
+
+from projektmatch import stages
+from projektmatch.config import Cfg
+
+
+class PMNotify(Component):
+    display_name = "PM 6 Notify"
+    description = "Benachrichtigungs-Mail bei created + Langfuse-Trace"
+    inputs = [
+        DataInput(name="ctx_in", display_name="Context"),
+        SecretStrInput(name="imap_password", display_name="Mail Password",
+                       value="PM_IMAP_PASSWORD", load_from_db=True,
+                       advanced=True),
+        MessageTextInput(name="notify_to", display_name="Notify To",
+                         value="PM_NOTIFY_TO", load_from_db=True,
+                         advanced=True),
+    ]
+    outputs = [Output(name="out", display_name="Result", method="build_out")]
+
+    def build_out(self) -> Message:
+        cfg = Cfg(imap_password=self.imap_password, notify_to=self.notify_to)
+        # stage_notify is called DIRECTLY (not via run_stage): it must run for
+        # every terminal status so failed runs still post their Langfuse trace.
+        ctx = stages.stage_notify(dict(self.ctx_in.data), cfg)
+        self.status = ctx.get("status", "")
+        return Message(text=stages.summary(ctx))

projekt-matching/components/pm_rules.py

@@ -0,0 +1,31 @@
+import sys
+
+if "/app/langflow" not in sys.path:
+    sys.path.insert(0, "/app/langflow")
+
+from langflow.custom import Component
+from langflow.io import DataInput, MessageTextInput, Output
+from langflow.schema import Data
+
+from projektmatch import stages
+from projektmatch.config import Cfg
+
+
+class PMRules(Component):
+    display_name = "PM 4 Rules+Gate"
+    description = ("Deterministisch: Misc-Regeln, Match-Berechnung, "
+                   "Beschreibungs-Markdown, Gate > Schwellwert")
+    inputs = [
+        DataInput(name="ctx_in", display_name="Context"),
+        MessageTextInput(name="threshold", display_name="Threshold",
+                         value="PM_THRESHOLD", load_from_db=True,
+                         advanced=True),
+    ]
+    outputs = [Output(name="out", display_name="Context", method="build_out")]
+
+    def build_out(self) -> Data:
+        cfg = Cfg(threshold=int(self.threshold))
+        ctx = stages.run_stage("rules", stages.stage_rules,
+                               dict(self.ctx_in.data), cfg)
+        self.status = f"{ctx.get('decision')} ({ctx.get('mustMatch')} %)"
+        return Data(data=ctx)

projekt-matching/deploy/build_flows.py

@@ -0,0 +1,331 @@
+#!/usr/bin/env python3
+"""Create Langflow global variables, API key, and both PM flows via API.
+
+Idempotent: variables are upserted; existing flows with the same name are
+deleted and recreated. Run AFTER deploy_files.sh. Needs secrets.local.env
+sourced into the environment (PM_IMAP_PASSWORD, PM_ESPO_API_KEY,
+PM_ESPO_BASE).
+
+Node/edge shapes and the two builtin_template category keys below were
+verified live against Langflow 1.10.0's GET /api/v1/all and the existing
+"vLLM Smoke Test" flow's edge dump (see task-10-report.md for the raw
+evidence). Notable deviations from the naive assumption:
+  - Webhook lives in the "input_output" category, not "data".
+  - Every custom Component's Data-typed `out` output reports its Langflow
+    wire type as "JSON" (not "Data") in outputs[].types.
+  - DataInput/JSONInput fields (`ctx_in`, `trigger`) report input_types as
+    the full ["Data", "JSON"], not just ["Data"].
+Edges are built from the live template's outputs[].types /
+template[field].input_types instead of hardcoded guesses, so any future
+Langflow version drift here is easy to re-derive by rerunning the Step-3
+probe in the task brief.
+
+IMPORTANT (found while debugging Step 6, see task-10-report.md): Langflow's
+`Component` base class defines `ctx` as a reserved @property (returns
+`self.graph.context`, the flow-level shared context store). The Task-9
+components originally named their inter-stage DataInput field "ctx", which
+silently shadows the framework property at attribute-read time -- the edge
+correctly delivers data into `self._attributes["ctx"]`, but `self.ctx`
+inside build_out() resolves to the *class property* instead (an empty
+dotdict), not the input. There is no error; the payload just silently comes
+back empty one hop downstream. Task 10 renamed the field to `ctx_in` in all
+five affected components (pm_extract.py, pm_match.py, pm_rules.py,
+pm_crm.py, pm_notify.py); the edges below target "ctx_in".
+"""
+import json
+import os
+import sys
+import time
+from pathlib import Path
+
+import requests
+
+BASE = "http://127.0.0.1:8090/api/v1"
+COMPONENTS = Path(__file__).resolve().parent.parent / "components"
+ENV_FILE = Path.home() / ".config" / "projekt-matching" / "env"
+
+GENERIC_VARS = {
+    "PM_ESPO_BASE": os.environ.get("PM_ESPO_BASE", ""),
+    "PM_NOTIFY_TO": "chancen@destengs.com",
+    "PM_ALERT_TO": "chancen@destengs.com",
+    "PM_THRESHOLD": "85",
+    "PM_VLLM_BASE": "http://host.containers.internal:8081/v1",
+    "PM_VLLM_MODEL": "AxionML/Qwen3.5-9B-NVFP4",
+    "PM_FLOW2_ID": "placeholder",
+}
+SECRET_VARS = ("PM_IMAP_PASSWORD", "PM_ESPO_API_KEY")
+
+
+def login():
+    r = requests.get(f"{BASE}/auto_login", timeout=15)
+    r.raise_for_status()
+    token = r.json()["access_token"]
+    return {"Authorization": f"Bearer {token}"}
+
+
+def upsert_variable(hdr, name, value, vtype):
+    existing = {v["name"]: v["id"] for v in
+                requests.get(f"{BASE}/variables/", headers=hdr,
+                             timeout=15).json()}
+    body = {"name": name, "value": value, "type": vtype,
+            "default_fields": []}
+    if name in existing:
+        requests.patch(f"{BASE}/variables/{existing[name]}", headers=hdr,
+                       json={"id": existing[name], "name": name,
+                             "value": value}, timeout=15).raise_for_status()
+    else:
+        requests.post(f"{BASE}/variables/", headers=hdr, json=body,
+                      timeout=15).raise_for_status()
+
+
+def get_api_key(hdr):
+    # Idempotent delete+recreate: the key VALUE is only returned at creation
+    # time, so reuse is impossible -- instead remove any previous
+    # "projekt-matching" keys (each earlier run left one behind otherwise)
+    # and mint a fresh one. Callers then refresh both the env file and the
+    # PM_LANGFLOW_API_KEY Credential variable, so nothing still references
+    # the deleted keys. Live API shape (Langflow 1.10.0): GET /api_key/
+    # returns {"total_count", "user_id", "api_keys": [{"id", "name", ...}]};
+    # DELETE /api_key/{id} removes one key.
+    r = requests.get(f"{BASE}/api_key/", headers=hdr, timeout=15)
+    r.raise_for_status()
+    for key in r.json().get("api_keys", []):
+        if key["name"] == "projekt-matching":
+            requests.delete(f"{BASE}/api_key/{key['id']}", headers=hdr,
+                            timeout=15).raise_for_status()
+    r = requests.post(f"{BASE}/api_key/", headers=hdr,
+                      json={"name": "projekt-matching"}, timeout=15)
+    r.raise_for_status()
+    return r.json()["api_key"]
+
+
+def component_template(hdr, code):
+    r = requests.post(f"{BASE}/custom_component", headers=hdr,
+                      json={"code": code}, timeout=60)
+    r.raise_for_status()
+    out = r.json()
+    return out["data"], out["type"]
+
+
+def builtin_template(hdr, category, name):
+    r = requests.get(f"{BASE}/all", headers=hdr, timeout=60)
+    r.raise_for_status()
+    return r.json()[category][name]
+
+
+def handle_str(d):
+    return json.dumps(d, separators=(",", ":")).replace('"', "œ")
+
+
+def make_node(node_id, comp_type, template, x):
+    return {"id": node_id, "type": "genericNode",
+            "position": {"x": x, "y": 0},
+            "data": {"type": comp_type, "id": node_id, "node": template,
+                     "showNode": True}}
+
+
+def out_spec(template, out_name):
+    """Look up an output's live wire types from a fetched node template."""
+    for o in template["outputs"]:
+        if o["name"] == out_name:
+            return o["types"]
+    raise KeyError(f"output {out_name!r} not found")
+
+
+def in_spec(template, field_name):
+    """Look up a field's live (input_types, type) from a node template."""
+    spec = template["template"][field_name]
+    return spec["input_types"], spec["type"]
+
+
+def make_edge(src_id, src_type, out_name, out_types, tgt_id, field,
+              input_types, field_type):
+    sh = {"dataType": src_type, "id": src_id, "name": out_name,
+          "output_types": out_types}
+    th = {"fieldName": field, "id": tgt_id, "inputTypes": input_types,
+          "type": field_type}
+    return {"id": f"reactflow__edge-{src_id}-{tgt_id}",
+            "source": src_id, "target": tgt_id,
+            "sourceHandle": handle_str(sh), "targetHandle": handle_str(th),
+            "data": {"sourceHandle": sh, "targetHandle": th},
+            "className": ""}
+
+
+def edge_between(src_id, src_type, src_tpl, out_name,
+                  tgt_id, tgt_tpl, field):
+    out_types = out_spec(src_tpl, out_name)
+    input_types, field_type = in_spec(tgt_tpl, field)
+    return make_edge(src_id, src_type, out_name, out_types,
+                      tgt_id, field, input_types, field_type)
+
+
+def _matches_flow(f, name, endpoint):
+    fname, fep = f["name"], f.get("endpoint_name") or ""
+    return (fname == name or fname.startswith(f"{name} (")
+            or fep == endpoint
+            or (fep.startswith(f"{endpoint}-")
+                and fep[len(endpoint) + 1:].isdigit()))
+
+
+def create_flow(hdr, name, endpoint, nodes, edges):
+    # Delete every previous incarnation. Langflow auto-renames on collision
+    # ("Name (1)", endpoint "endpoint-1") instead of rejecting, so a plain
+    # exact-name match leaves suffixed zombies behind that then keep the
+    # endpoint slot occupied on the next run -- match the auto-suffixed
+    # variants (by name AND endpoint) too, and fail loudly if a delete
+    # doesn't stick. Then wait until the listing no longer shows any of
+    # them: the create-time uniqueness check otherwise still sees the
+    # just-deleted rows and suffixes the new flow anyway (observed live
+    # on 1.10.0).
+    flows = requests.get(f"{BASE}/flows/", headers=hdr, timeout=30).json()
+    for f in flows:
+        if _matches_flow(f, name, endpoint):
+            requests.delete(f"{BASE}/flows/{f['id']}", headers=hdr,
+                            timeout=15).raise_for_status()
+    for _ in range(20):
+        flows = requests.get(f"{BASE}/flows/", headers=hdr,
+                             timeout=30).json()
+        if not any(_matches_flow(f, name, endpoint) for f in flows):
+            break
+        time.sleep(0.5)
+
+    body = {"name": name, "endpoint_name": endpoint, "is_component": False,
+            "data": {"nodes": nodes, "edges": edges,
+                     "viewport": {"x": 0, "y": 0, "zoom": 0.5}}}
+    r = requests.post(f"{BASE}/flows/", headers=hdr, json=body, timeout=30)
+    r.raise_for_status()
+    created = r.json()
+
+    # Belt-and-braces: if the name/endpoint still got suffixed, PATCH them
+    # back. The PATCH can 404 transiently right after the POST (read-your-
+    # writes lag observed live), so retry briefly.
+    fixes = {}
+    if created["name"] != name:
+        fixes["name"] = name
+    if (created.get("endpoint_name") or "") != endpoint:
+        fixes["endpoint_name"] = endpoint
+    if fixes:
+        for attempt in range(10):
+            pr = requests.patch(f"{BASE}/flows/{created['id']}",
+                                headers=hdr, json=fixes, timeout=15)
+            if pr.status_code == 404:
+                time.sleep(1)
+                continue
+            pr.raise_for_status()
+            break
+        else:
+            raise RuntimeError(
+                f"could not normalize flow {created['id']} to "
+                f"name={name!r} endpoint={endpoint!r}")
+    return created["id"]
+
+
+def main():
+    hdr = login()
+    for name in SECRET_VARS:
+        value = os.environ.get(name, "")
+        if not value:
+            # Langflow rejects empty Credential values outright ("Variable
+            # value cannot be empty"). PM_ESPO_API_KEY legitimately comes
+            # back empty here when secrets.local.env's grep against
+            # /home/tlg/mkt/bewerb/.secrets/espocrm-api.md can't read that
+            # chmod-600 file (see task-10-report.md). Use an obviously-fake
+            # placeholder so the deploy can proceed -- stage_crm() only
+            # calls espocrm with this value when ctx["decision"] ==
+            # "consider", so the reject-path smoke test (Step 6) is
+            # unaffected; the accept-path / real CRM writes remain
+            # unverified until a real key is supplied.
+            value = f"UNSET-{name}-blocked-by-file-permissions"
+            print(f"WARN: {name} is empty in the environment, using "
+                  f"placeholder credential value", file=sys.stderr)
+        upsert_variable(hdr, name, value, "Credential")
+    for name, value in GENERIC_VARS.items():
+        if name == "PM_FLOW2_ID":
+            continue
+        if not value:
+            # Same empty-value rejection as above, for PM_ESPO_BASE.
+            value = f"UNSET-{name}-blocked-by-file-permissions"
+            print(f"WARN: {name} is empty in the environment, using "
+                  f"placeholder value", file=sys.stderr)
+        upsert_variable(hdr, name, value, "Generic")
+    api_key = get_api_key(hdr)
+    # PMIngest reads the run-API key via load_from_db -> must exist as a
+    # Credential global variable, not only in the host env file.
+    upsert_variable(hdr, "PM_LANGFLOW_API_KEY", api_key, "Credential")
+
+    def custom(fname):
+        code = (COMPONENTS / fname).read_text()
+        return component_template(hdr, code)
+
+    # ---- Flow 2: TextInput -> PMFetch -> ... -> PMNotify -> TextOutput
+    text_in = builtin_template(hdr, "input_output", "TextInput")
+    text_out = builtin_template(hdr, "input_output", "TextOutput")
+    chain = [("TextInput-pm2i", "TextInput", text_in)]
+    for fname, nid in [("pm_fetch.py", "PMFetch-pm2a"),
+                       ("pm_extract.py", "PMExtract-pm2b"),
+                       ("pm_match.py", "PMMatch-pm2c"),
+                       ("pm_rules.py", "PMRules-pm2d"),
+                       ("pm_crm.py", "PMCrm-pm2e"),
+                       ("pm_notify.py", "PMNotify-pm2f")]:
+        tpl, ctype = custom(fname)
+        chain.append((nid, ctype, tpl))
+    chain.append(("TextOutput-pm2o", "TextOutput", text_out))
+    nodes2 = [make_node(nid, ctype, tpl, x=i * 380)
+              for i, (nid, ctype, tpl) in enumerate(chain)]
+
+    by_id = {nid: tpl for nid, _, tpl in chain}
+    by_type = {nid: ctype for nid, ctype, _ in chain}
+
+    edges2 = [
+        edge_between("TextInput-pm2i", "TextInput", by_id["TextInput-pm2i"],
+                     "text", "PMFetch-pm2a", by_id["PMFetch-pm2a"],
+                     "payload"),
+        edge_between("PMFetch-pm2a", by_type["PMFetch-pm2a"],
+                     by_id["PMFetch-pm2a"], "out",
+                     "PMExtract-pm2b", by_id["PMExtract-pm2b"], "ctx_in"),
+        edge_between("PMExtract-pm2b", by_type["PMExtract-pm2b"],
+                     by_id["PMExtract-pm2b"], "out",
+                     "PMMatch-pm2c", by_id["PMMatch-pm2c"], "ctx_in"),
+        edge_between("PMMatch-pm2c", by_type["PMMatch-pm2c"],
+                     by_id["PMMatch-pm2c"], "out",
+                     "PMRules-pm2d", by_id["PMRules-pm2d"], "ctx_in"),
+        edge_between("PMRules-pm2d", by_type["PMRules-pm2d"],
+                     by_id["PMRules-pm2d"], "out",
+                     "PMCrm-pm2e", by_id["PMCrm-pm2e"], "ctx_in"),
+        edge_between("PMCrm-pm2e", by_type["PMCrm-pm2e"],
+                     by_id["PMCrm-pm2e"], "out",
+                     "PMNotify-pm2f", by_id["PMNotify-pm2f"], "ctx_in"),
+        edge_between("PMNotify-pm2f", by_type["PMNotify-pm2f"],
+                     by_id["PMNotify-pm2f"], "out",
+                     "TextOutput-pm2o", by_id["TextOutput-pm2o"],
+                     "input_value"),
+    ]
+    flow2_id = create_flow(hdr, "PM Projekt bewerten", "pm-flow2",
+                           nodes2, edges2)
+    upsert_variable(hdr, "PM_FLOW2_ID", flow2_id, "Generic")
+
+    # ---- Flow 1: Webhook -> PMIngest -> TextOutput
+    webhook = builtin_template(hdr, "input_output", "Webhook")
+    tpl_ing, ctype_ing = custom("pm_ingest.py")
+    nodes1 = [make_node("Webhook-pm1w", "Webhook", webhook, 0),
+              make_node("PMIngest-pm1a", ctype_ing, tpl_ing, 380),
+              make_node("TextOutput-pm1o", "TextOutput", text_out, 760)]
+    edges1 = [
+        edge_between("Webhook-pm1w", "Webhook", webhook, "output_data",
+                     "PMIngest-pm1a", tpl_ing, "trigger"),
+        edge_between("PMIngest-pm1a", ctype_ing, tpl_ing, "out",
+                     "TextOutput-pm1o", text_out, "input_value"),
+    ]
+    flow1_id = create_flow(hdr, "PM Ingest", "pm-ingest", nodes1, edges1)
+
+    ENV_FILE.parent.mkdir(parents=True, exist_ok=True)
+    ENV_FILE.write_text(
+        f"LANGFLOW_API_KEY={api_key}\n"
+        f"FLOW1_ID={flow1_id}\nFLOW2_ID={flow2_id}\n"
+        f"FLOW1_WEBHOOK=http://127.0.0.1:8090/api/v1/webhook/pm-ingest\n")
+    ENV_FILE.chmod(0o600)
+    print(f"flow1={flow1_id} flow2={flow2_id}\nenv -> {ENV_FILE}")
+
+
+if __name__ == "__main__":
+    sys.exit(main())

projekt-matching/deploy/deploy_files.sh

@@ -0,0 +1,71 @@
+#!/bin/bash
+# Copy the projektmatch package and vorgaben files into the Langflow data dir
+# (mounted at /app/langflow in the container). Idempotent.
+#
+# NOTE: /home/tlg/mkt/bewerb/vorgaben/Lebenslauf_Dr-Ing_Thomas_Langer.md and
+# /home/tlg/mkt/bewerb/.secrets/espocrm-api.md are chmod 600, owned by tlg.
+# This script normally runs as the lwc automation user, which cannot read
+# those two files (verified: `cp` fails with EACCES). Rather than aborting
+# the whole deploy (set -e) on a permission error, each vorgaben file is
+# copied individually and a missing/unreadable file only produces a WARN,
+# so the parts of the pipeline that don't need it still deploy correctly.
+set -e
+SRC="$(cd "$(dirname "$0")/.." && pwd)"
+DEST="$HOME/.local/share/langflow_pod/langflow-data"
+
+rsync -a --delete --exclude __pycache__ "$SRC/projektmatch/" "$DEST/projektmatch/"
+mkdir -p "$DEST/vorgaben"
+for f in /home/tlg/mkt/bewerb/vorgaben/Lebenslauf_Dr-Ing_Thomas_Langer.md \
+         /home/tlg/mkt/bewerb/vorgaben/rahmenbedingungen.md; do
+  if [ -r "$f" ]; then
+    cp "$f" "$DEST/vorgaben/"
+  else
+    echo "WARN: cannot read $f as $(whoami) (permission denied) - skipping" >&2
+  fi
+done
+# Langflow runs as uid 1000 gid 0 -> needs group read
+chmod -R g+rX "$DEST/projektmatch" "$DEST/vorgaben"
+
+# /app/langflow is already on sys.path inside the container: create_pod_langflow.sh
+# runs the langflow container with `-e PYTHONPATH=/app/langflow` baked into the
+# `podman run` invocation itself (see the comment block above that env var in
+# create_pod_langflow.sh), so "import projektmatch" resolves from the very first
+# boot without any venv-local .pth file. Do NOT reintroduce a .pth here: this
+# container's systemd unit is generated with `podman generate systemd --new`
+# (--rm + --replace in ExecStart), so the venv's site-packages lives in the
+# container's ephemeral overlay and is wiped on every recreation anyway.
+#
+# The freshly copied code still needs a running-process restart to take effect
+# (Python doesn't hot-reload modules), so restart via the systemd user unit --
+# never a bare `podman restart`: since the unit is `--new`-managed, a bare
+# `podman restart` on a container systemd also manages can leave systemd's
+# view of the unit out of sync with the container's actual state (documented
+# pitfall: the unit can end up believing the container is REMOVED). Restarting
+# through systemd keeps systemd and podman in agreement.
+CTR="langflow_ctr"
+SVC="container-langflow_ctr.service"
+if podman inspect "$CTR" >/dev/null 2>&1; then
+  echo "Restarting $CTR via $SVC to pick up new code..."
+  if systemctl --user restart "$SVC"; then
+    code=""
+    for i in $(seq 1 40); do
+      code=$(curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8090/health || true)
+      [ "$code" = "200" ] && break
+      sleep 3
+    done
+    if [ "$code" != "200" ]; then
+      echo "WARN: $CTR did not come back healthy within 120s after restart" \
+           "(last health code: ${code:-none}). Recover with:" \
+           "'systemctl --user restart $SVC && curl -fsS http://127.0.0.1:8090/health'" >&2
+    fi
+  else
+    echo "WARN: 'systemctl --user restart $SVC' failed - $CTR may not have" \
+         "picked up the new code. Recover with:" \
+         "'systemctl --user restart $SVC'" >&2
+  fi
+else
+  echo "WARN: container $CTR not found - skipping restart" \
+       "(recover with: 'systemctl --user start $SVC')" >&2
+fi
+
+echo "Deployed to $DEST"

projekt-matching/deploy/projekt-matching.service

@@ -0,0 +1,6 @@
+[Unit]
+Description=Trigger PM Ingest Langflow webhook
+
+[Service]
+Type=oneshot
+ExecStart=%h/bin/projekt-matching/deploy/trigger_webhook.sh

projekt-matching/deploy/projekt-matching.timer

@@ -0,0 +1,9 @@
+[Unit]
+Description=Poll chancen@ inbox via PM Ingest every 5 minutes
+
+[Timer]
+OnCalendar=*:00/5
+Persistent=false
+
+[Install]
+WantedBy=timers.target

projekt-matching/deploy/setup_langfuse.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env python3
+"""Create Langfuse org/project/keys + score configs for projekt-matching.
+
+Uses the Langfuse v3.195 API. IMPORTANT deviation from the original plan:
+the Organization Management API (Bearer ADMIN_API_KEY, /api/admin/...) is an
+Enterprise-only feature in this self-hosted OSS build -- confirmed live via
+`curl -H "Authorization: Bearer $ADMIN_API_KEY" .../api/admin/organizations`
+-> 403 {"error":"This feature is not available on your current plan."}. The
+route exists (not a 404 / wrong-URL guess), it is just plan-gated.
+
+Fallback used instead (still API-driven, no browser): log in as the
+bootstrap admin user (LANGFUSE_INIT_USER_EMAIL/PASSWORD from the pod
+script) via the NextAuth credentials flow to get a session cookie, then call
+the same internal tRPC endpoints the Langfuse web UI itself uses
+(organizations.create, projects.create, projectApiKeys.create) to create the
+org, project and a project API key pair. Score configs ARE available on the
+public, documented REST API (POST /api/public/score-configs) once we have a
+project key pair, so those use requests+Basic auth like the public API
+elsewhere in this repo.
+
+Idempotent: reuses an existing "projekt-matching-org"/"projekt-matching" org
++ project (looked up via the session endpoint) instead of erroring out, and
+score-config creation is safe to retry (Langfuse allows same-name configs;
+re-running this script when configs already exist will just add duplicates,
+so check the UI/API before rerunning after a successful first run).
+
+Prints the project keys; Step 3 (create_pod_langflow.sh) wires them into the
+Langflow container env."""
+from __future__ import annotations
+
+import re
+import sys
+
+import requests
+
+BASE = "http://127.0.0.1:8091"
+POD_SCRIPT = "/home/lwc/bin/create_pod_langflow.sh"
+
+
+def _pod_var(name: str) -> str:
+    """Read a VAR="value" assignment from the pod script (single source of
+    configuration -- avoids stale duplicated literals here if the pod
+    script rotates its credentials)."""
+    with open(POD_SCRIPT, encoding="utf-8") as f:
+        text = f.read()
+    m = re.search(rf'^{name}="([^"]+)"', text, re.M)
+    if not m:
+        sys.exit(f"{name} not found in {POD_SCRIPT}")
+    return m.group(1)
+
+
+ADMIN_BEARER = {
+    "Authorization": "Bearer " + _pod_var("LANGFUSE_ADMIN_API_KEY")}
+LOGIN_EMAIL = _pod_var("LANGFUSE_INIT_EMAIL")
+LOGIN_PASSWORD = _pod_var("LANGFUSE_INIT_PASSWORD")
+ORG_NAME = "projekt-matching-org"
+PROJECT_NAME = "projekt-matching"
+SCORE_CONFIGS = (
+    ("extraction-correct",
+     "Must/Nice/Misc korrekt aus der Ausschreibung abgeleitet?"),
+    ("matching-correct",
+     "✅/❌/❔-Bewertungen gegen den Lebenslauf korrekt?"),
+)
+SCORE_CATEGORIES = [
+    {"label": "correct", "value": 1},
+    {"label": "partially-correct", "value": 0.5},
+    {"label": "wrong", "value": 0},
+]
+
+
+def try_admin_api() -> dict | None:
+    """Attempt the documented Organization Management API. Returns the
+    project key pair dict on success, None if the feature is unavailable
+    (403) so the caller can fall back."""
+    r = requests.post(f"{BASE}/api/admin/organizations",
+                       json={"name": ORG_NAME}, headers=ADMIN_BEARER,
+                       timeout=30)
+    if r.status_code == 403:
+        print("Admin Organization Management API is plan-gated (403) on "
+              "this self-hosted OSS build -- falling back to the session "
+              "+ tRPC method.", file=sys.stderr)
+        return None
+    if r.status_code == 404:
+        print("Admin API route missing (404) -- falling back to the "
+              "session + tRPC method.", file=sys.stderr)
+        return None
+    r.raise_for_status()
+    org = r.json()
+    org_key = requests.post(
+        f"{BASE}/api/admin/organizations/{org['id']}/apiKeys", json={},
+        headers=ADMIN_BEARER, timeout=30)
+    org_key.raise_for_status()
+    ok = org_key.json()
+    org_auth = (ok["publicKey"], ok["secretKey"])
+    proj = requests.post(f"{BASE}/api/public/projects",
+                          json={"name": PROJECT_NAME, "retention": 0},
+                          auth=org_auth, timeout=30)
+    proj.raise_for_status()
+    project = proj.json()
+    keys = requests.post(
+        f"{BASE}/api/public/projects/{project['id']}/apiKeys", json={},
+        auth=org_auth, timeout=30)
+    keys.raise_for_status()
+    return keys.json()
+
+
+def session_login() -> tuple[requests.Session, dict]:
+    s = requests.Session()
+    csrf = s.get(f"{BASE}/api/auth/csrf", timeout=15).json()["csrfToken"]
+    r = s.post(f"{BASE}/api/auth/callback/credentials", data={
+        "email": LOGIN_EMAIL, "password": LOGIN_PASSWORD, "csrfToken": csrf,
+        "callbackUrl": f"{BASE}/", "json": "true",
+    }, timeout=15)
+    r.raise_for_status()
+    sess = s.get(f"{BASE}/api/auth/session", timeout=15).json()
+    if not sess.get("user"):
+        sys.exit(f"Login failed: {sess}")
+    return s, sess
+
+
+def trpc(s: requests.Session, path: str, payload: dict) -> dict:
+    r = s.post(f"{BASE}/api/trpc/{path}", json={"json": payload},
+               timeout=30)
+    if r.status_code >= 400:
+        sys.exit(f"tRPC {path} failed ({r.status_code}): {r.text[:500]}")
+    return r.json()["result"]["data"]["json"]
+
+
+def via_session_trpc() -> dict:
+    s, sess = session_login()
+    org_id = project_id = None
+    for org in sess["user"]["organizations"]:
+        if org["name"] == ORG_NAME or org["id"] == ORG_NAME:
+            org_id = org["id"]
+            for proj in org["projects"]:
+                if proj["name"] == PROJECT_NAME:
+                    project_id = proj["id"]
+                    break
+            break
+    if org_id is None:
+        org = trpc(s, "organizations.create", {"name": ORG_NAME})
+        org_id = org["id"]
+        print(f"Created organization {ORG_NAME} ({org_id})", file=sys.stderr)
+    else:
+        print(f"Reusing existing organization {ORG_NAME} ({org_id})",
+              file=sys.stderr)
+    if project_id is None:
+        proj = trpc(s, "projects.create",
+                    {"orgId": org_id, "name": PROJECT_NAME})
+        project_id = proj["id"]
+        print(f"Created project {PROJECT_NAME} ({project_id})",
+              file=sys.stderr)
+    else:
+        print(f"Reusing existing project {PROJECT_NAME} ({project_id})",
+              file=sys.stderr)
+    keys = trpc(s, "projectApiKeys.create", {"projectId": project_id})
+    return keys
+
+
+def create_score_configs(pk: str, sk: str) -> None:
+    for name, desc in SCORE_CONFIGS:
+        r = requests.post(f"{BASE}/api/public/score-configs", auth=(pk, sk),
+                           json={"name": name, "dataType": "CATEGORICAL",
+                                 "description": desc,
+                                 "categories": SCORE_CATEGORIES},
+                           timeout=30)
+        r.raise_for_status()
+        print(f"Score config '{name}' created: {r.json()['id']}",
+              file=sys.stderr)
+
+
+def main() -> None:
+    keys = try_admin_api()
+    if keys is None:
+        keys = via_session_trpc()
+    pk, sk = keys["publicKey"], keys["secretKey"]
+    create_score_configs(pk, sk)
+    print(f"PM_LANGFUSE_PUBLIC_KEY={pk}")
+    print(f"PM_LANGFUSE_SECRET_KEY={sk}")
+
+
+if __name__ == "__main__":
+    main()

projekt-matching/deploy/trigger_webhook.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+# Fire the PM Ingest webhook. Carries no logic; failures are harmless
+# (state lives in the IMAP inbox, next tick retries).
+set -u
+source "$HOME/.config/projekt-matching/env"
+curl -sS -m 15 -X POST "$FLOW1_WEBHOOK" \
+  -H "x-api-key: $LANGFLOW_API_KEY" -H 'Content-Type: application/json' \
+  -d '{"source": "systemd-timer"}' || true
+echo

projekt-matching/projektmatch/__init__.py

@@ -0,0 +1 @@
+"""projektmatch — automated freelancermap project matching (Langflow backend)."""

projekt-matching/projektmatch/config.py

@@ -0,0 +1,20 @@
+"""Runtime configuration passed from Langflow component inputs."""
+from dataclasses import dataclass
+
+
+@dataclass
+class Cfg:
+    imap_user: str = "chancen@destengs.com"
+    imap_password: str = ""
+    espo_base: str = ""
+    espo_api_key: str = ""
+    notify_to: str = ""
+    alert_to: str = ""
+    threshold: int = 85
+    vllm_base: str = ""
+    vllm_model: str = ""
+    flow2_id: str = ""
+    langflow_api_key: str = ""
+    langflow_base: str = "http://127.0.0.1:7860"
+    data_dir: str = "/app/langflow"
+    crm_web_base: str = "https://crm.creature-go.com"

projekt-matching/projektmatch/espocrm.py

@@ -0,0 +1,108 @@
+"""Minimal EspoCRM REST client (X-Api-Key auth, in-run retries)."""
+from __future__ import annotations
+
+import re
+import time
+
+import requests
+
+
+class EspoError(RuntimeError):
+    pass
+
+
+class EspoClient:
+    def __init__(self, base: str, api_key: str, retries: int = 2,
+                 backoff: float = 2.0):
+        self.base = base.rstrip("/")
+        self.session = requests.Session()
+        self.session.headers["X-Api-Key"] = api_key
+        self.retries = retries
+        self.backoff = backoff
+
+    def _req(self, method: str, path: str, **kw):
+        last = None
+        for attempt in range(self.retries + 1):
+            try:
+                r = self.session.request(method, f"{self.base}/{path}",
+                                         timeout=30, **kw)
+                if r.status_code < 500:
+                    if r.status_code >= 400:
+                        raise EspoError(
+                            f"{method} {path} -> {r.status_code} "
+                            f"{r.headers.get('X-Status-Reason', '')}")
+                    return r.json() if r.text else {}
+                last = EspoError(f"{method} {path} -> {r.status_code}")
+            except (requests.ConnectionError, requests.Timeout) as exc:
+                last = exc
+            time.sleep(self.backoff * (attempt + 1))
+        raise EspoError(str(last))
+
+    def search(self, entity, wtype, attr, value, select="name", max_size=50):
+        params = {"where[0][type]": wtype, "where[0][attribute]": attr,
+                  "where[0][value]": value, "select": select,
+                  "maxSize": max_size}
+        return self._req("GET", entity, params=params).get("list", [])
+
+    def find_opportunity_by_link(self, url):
+        hits = self.search("Opportunity", "equals", "cProjektlink", url)
+        return hits[0] if hits else None
+
+    def team_id(self, name):
+        hits = self.search("Team", "equals", "name", name)
+        if not hits:
+            raise EspoError(
+                f"Team '{name}' nicht gefunden — cowork-api braucht Lese-"
+                f"Zugriff auf Team, und das Team muss existieren")
+        return hits[0]["id"]
+
+    def ensure_account(self, name, acc_type):
+        for hit in self.search("Account", "contains", "name",
+                               core_token(name), select="name,type"):
+            if hit["name"].strip().lower() == name.strip().lower():
+                return hit["id"]
+        return self._req("POST", "Account",
+                         json={"name": name, "type": acc_type})["id"]
+
+    def ensure_contact(self, first, last, account_id):
+        full = f"{first} {last}".strip().lower()
+        for hit in self.search("Contact", "contains", "name", last,
+                               select="name,accountName"):
+            if hit["name"].strip().lower() == full:
+                return hit["id"]
+        return self._req("POST", "Contact",
+                         json={"firstName": first, "lastName": last,
+                               "accountId": account_id})["id"]
+
+    def unique_opportunity_name(self, name):
+        existing = {h["name"] for h in self.search(
+            "Opportunity", "startsWith", "name", name, max_size=100)}
+        if name not in existing:
+            return name
+        n = 2
+        while f"{name} ({n})" in existing:
+            n += 1
+        return f"{name} ({n})"
+
+    def create_opportunity(self, payload):
+        return self._req("POST", "Opportunity", json=payload)
+
+    def get_opportunity(self, oid):
+        return self._req("GET", f"Opportunity/{oid}")
+
+    def delete(self, entity, oid):
+        return self._req("DELETE", f"{entity}/{oid}")
+
+
+def core_token(name: str) -> str:
+    for tok in re.split(r"[^\wÄÖÜäöüß]+", name or ""):
+        if len(tok) > 2:
+            return tok
+    return name
+
+
+def split_person(full: str):
+    parts = (full or "").split()
+    if not parts:
+        return ("", "")
+    return (" ".join(parts[:-1]), parts[-1])

projekt-matching/projektmatch/ingest.py

@@ -0,0 +1,142 @@
+"""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}}
+        try:
+            for uid in box.unchecked_uids():
+                try:
+                    msg = box.fetch(uid)
+                    subject = decode_subject(msg)
+                    html, text = mailparse.bodies(msg)
+                except Exception:              # noqa: BLE001
+                    box.flag_checked(uid)      # nie wieder anfassen, Pipeline läuft weiter
+                    summary["unreadable"] = summary.get("unreadable", 0) + 1
+                    continue
+                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))
+                    if result.get("status") not in SUMMARY_KEYS:
+                        result.update(
+                            status="failed",
+                            error=result.get("error")
+                            or f"unbekannter Status: {result.get('status')!r}")
+                    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)
+                    try:
+                        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")
+                    except Exception:          # noqa: BLE001
+                        summary["alertFailed"] = summary.get("alertFailed", 0) + 1
+                box.move_to_trash(uid)
+            return summary
+        finally:
+            box.close()
+    finally:
+        release_lock(lock)

projekt-matching/projektmatch/llm.py

@@ -0,0 +1,197 @@
+"""vLLM (OpenAI-compatible) calls with response_format json_schema + German prompts.
+
+max_tokens stays UNSET so nothing can truncate the final answer (65k
+context window). Thinking is DISABLED via chat_template_kwargs: with the
+vLLM qwen3 reasoning parser active, json_schema guided decoding plus
+thinking degenerates — the run burns the entire context window
+(finish_reason "length", ~63k completion tokens) and returns empty or
+truncated content (verified live in Task 13). Without thinking the same
+extract call answers in ~35 s with valid schema-conformant JSON.
+"""
+from __future__ import annotations
+
+import json
+import re
+import time
+
+import requests
+
+THINK_RE = re.compile(r"<think>.*?</think>", re.S)
+PAGE_TEXT_LIMIT = 24000
+
+
+class LlmError(RuntimeError):
+    pass
+
+
+EXTRACT_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "projectName": {"type": "string"},
+        "offerType": {"enum": ["Projekt", "Arbeitnehmer-Angebot", "ANÜ"]},
+        "buyerType": {"enum": ["agency", "direct"]},
+        "companyName": {"type": ["string", "null"]},
+        "contactPerson": {"type": ["string", "null"]},
+        "requirements": {
+            "type": "array",
+            "items": {
+                "type": "object",
+                "properties": {
+                    "text": {"type": "string"},
+                    "kat": {"enum": ["Must", "Nice", "Misc"]},
+                    "miscType": {"enum": ["start", "workload", "duration",
+                                           "location", "security", "other"]},
+                    "startDate": {"type": ["string", "null"]},
+                    "workloadPercent": {"type": ["integer", "null"]},
+                    "remotePercent": {"type": ["integer", "null"]},
+                    "onsiteLocation": {"type": ["string", "null"]},
+                },
+                "required": ["text", "kat", "miscType", "startDate",
+                             "workloadPercent", "remotePercent",
+                             "onsiteLocation"],
+            },
+        },
+    },
+    "required": ["projectName", "offerType", "buyerType", "companyName",
+                 "contactPerson", "requirements"],
+}
+
+MATCH_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "ratings": {
+            "type": "array",
+            "items": {
+                "type": "object",
+                "properties": {
+                    "nr": {"type": "integer"},
+                    "rating": {"enum": ["yes", "no", "unknown"]},
+                    "reason": {"type": "string"},
+                },
+                "required": ["nr", "rating", "reason"],
+            },
+        },
+    },
+    "required": ["ratings"],
+}
+
+EXTRACT_SYSTEM = """Du extrahierst Anforderungen aus deutschen \
+Freiberufler-Projektausschreibungen. Antworte NUR mit JSON nach Schema.
+
+Regeln:
+- projectName: Titel der Ausschreibung OHNE Portal-Zusatz (z. B. ohne "auf \
+www.freelancermap.de").
+- requirements: jede Anforderung einzeln, im Originalwortlaut (behutsames \
+Kürzen erlaubt, Bedeutung nie verändern). Rahmenbedingungen (Start, \
+Einsatzort/Remote-Anteil, Auslastung, Laufzeit) sind Anforderungen der \
+Kategorie Misc.
+- kat: Must = zwingend (Abschnitt "Must-haves"/"Anforderungen"; "zwingend", \
+"erforderlich", "vorausgesetzt", "sehr gute Kenntnisse"). Nice = optional \
+(Abschnitt "Nice-to-haves"; "von Vorteil", "wünschenswert", "idealerweise", \
+"plus"). Misc = Rahmenbedingungen und alles, was weder Muss noch \
+Wunsch-Qualifikation ist. Explizite Abschnittsüberschriften haben Vorrang \
+vor Signalwörtern; "idealerweise" INNERHALB einer Must-Zeile lässt die \
+Zeile Must bleiben. Nice-Signalwörter gelten NUR für die Zeile, in der \
+sie selbst stehen — sie färben NIE auf folgende Zeilen ab: Jede Zeile \
+eines Anforderungs-Abschnitts OHNE eigenes Nice-Signalwort ist Must, \
+auch wenn direkt davor Nice-Zeilen stehen — AUSNAHME: Zeilen unter einer \
+expliziten Nice-to-haves-Überschrift bleiben Nice. Beispiel: Auf \
+"Idealerweise zusätzlich Erfahrung mit X" (Nice) folgt "Erfahrung mit Y" \
+ohne Signalwort → Y ist Must.
+- miscType nur für Misc-Zeilen relevant (sonst "other"): start = \
+Projektstart/Verfügbarkeit (startDate als ISO-Datum YYYY-MM-DD, wenn ein \
+konkretes Datum genannt ist, sonst null); workload = Auslastung \
+(workloadPercent 0-100 oder null); duration = Laufzeit; location = \
+Einsatzort/Remote (remotePercent 0-100 oder null; onsiteLocation = \
+Ortsname oder null); security = Sicherheitsüberprüfung (SÜ, SÜ1/SÜ2/SÜ3, \
+Ü2, Geheimschutz).
+- offerType: "Projekt" = Freiberufler-/Werkauftrag (auch über Agentur). \
+"Arbeitnehmer-Angebot" bei Festanstellung ("Festanstellung", "unbefristet", \
+"Gehalt", "Arbeitsvertrag"). "ANÜ" bei Arbeitnehmerüberlassung ("ANÜ", \
+"AÜG", "Überlassung", "Zeitarbeit"). Im Zweifel "Projekt".
+- buyerType: "agency" bei Personaldienstleistern/Vermittlern (Hays, \
+GULP/Randstad, SThree, Computer Futures, Aristo, freelancermap-Vermittler, \
+"im Auftrag unseres Kunden", "für unseren Kunden"), sonst "direct". Im \
+Zweifel "agency".
+- companyName: Name der Agentur bzw. des Endkunden, sonst null. \
+contactPerson: vollständiger Name der Ansprechperson, sonst null."""
+
+MATCH_SYSTEM = """Du bewertest nüchtern und streng, ob ein Lebenslauf \
+einzelne Projekt-Anforderungen abdeckt. Antworte NUR mit JSON nach Schema: \
+für JEDE übergebene Nr. genau ein Eintrag in ratings.
+
+Bewertung:
+- "yes" NUR bei klarer Evidenz im Lebenslauf.
+- "no" wenn der Lebenslauf nichts Belastbares hergibt. Streng bleiben — \
+eine geschönte Bewertung macht die Match-Zahlen wertlos. Kalibrierung: \
+Proof-of-Concept-Erfahrung deckt "produktiven Betrieb" NICHT ab; \
+"mehrjährig" wörtlich nehmen; ein Produktname (z. B. "Azure DevOps \
+Server") belegt KEINE Cloud-Plattform-Erfahrung.
+- "unknown" NUR bei echter Teilevidenz, wenn die Entscheidung von Wissen \
+abhängt, das nur der Kandidat selbst hat. Zusammengesetzte Anforderungen \
+("X sowie Y", "X und Y"): klare Evidenz für einen Teil und keine \
+Gegen-Evidenz für den anderen → "unknown", nicht "no".
+- "wie z. B."-Aufzählungen: gleichwertige Alternativen zählen als Abdeckung \
+(Beispiel: Ollama/llama.cpp/Transformers decken "LLM-Inference-Stacks wie \
+z. B. vLLM, TGI, Triton" ab).
+- reason: EIN kurzer deutscher Satz mit der Begründung."""
+
+
+def _post(base, model, messages, schema):
+    body = {"model": model, "messages": messages, "temperature": 0.1,
+            "chat_template_kwargs": {"enable_thinking": False},
+            "response_format": {"type": "json_schema", "json_schema": {
+                "name": "out", "schema": schema}}}
+    return requests.post(f"{base.rstrip('/')}/chat/completions", json=body,
+                         timeout=1500)
+
+
+def chat_json(base, model, messages, schema, attempts=3):
+    if attempts < 1:
+        raise ValueError("attempts must be >= 1")
+    last = None
+    for attempt in range(attempts):
+        resp = _post(base, model, messages, schema)
+        if resp.status_code != 200:
+            raise LlmError(f"vLLM HTTP {resp.status_code}: "
+                           f"{getattr(resp, 'text', '')[:300]}")
+        content = resp.json()["choices"][0]["message"]["content"] or ""
+        content = THINK_RE.sub("", content).strip()
+        if content.startswith("```"):
+            content = re.sub(r"^```[a-z]*\s*|\s*```$", "", content).strip()
+        try:
+            return json.loads(content)
+        except json.JSONDecodeError as exc:
+            last = LlmError(f"LLM lieferte kein JSON: {exc}: {content[:200]}")
+            time.sleep(10 * (attempt + 1))
+    raise last
+
+
+def extract_project(base, model, page_text):
+    messages = [{"role": "system", "content": EXTRACT_SYSTEM},
+                {"role": "user", "content":
+                 "Ausschreibungstext:\n\n" + page_text[:PAGE_TEXT_LIMIT]}]
+    out = chat_json(base, model, messages, EXTRACT_SCHEMA)
+    if not out.get("requirements"):
+        raise LlmError("Extraktion ohne Anforderungen")
+    return out
+
+
+def match_cv(base, model, cv_text, requirements):
+    listing = "\n".join(f"{r['nr']}. {r['text']}" for r in requirements)
+    messages = [{"role": "system", "content": MATCH_SYSTEM},
+                {"role": "user", "content":
+                 f"Lebenslauf:\n\n{cv_text}\n\nAnforderungen:\n{listing}"}]
+    wanted = {r["nr"] for r in requirements}
+    for attempt in range(2):
+        out = chat_json(base, model, messages, MATCH_SCHEMA)
+        got = {r["nr"]: r for r in out.get("ratings", []) if r["nr"] in wanted}
+        if set(got) == wanted:
+            return [got[n] for n in sorted(got)]
+        missing = sorted(wanted - set(got))
+        messages = messages + [
+            {"role": "assistant", "content": json.dumps(out)},
+            {"role": "user", "content":
+             f"Es fehlen Bewertungen für Nr. {missing}. Antworte erneut "
+             f"mit ratings für ALLE Nummern."}]
+    raise LlmError(f"Matching unvollständig, fehlend: {missing}")

projekt-matching/projektmatch/mailer.py

@@ -0,0 +1,83 @@
+"""IMAP mailbox handling and SMTP sending for chancen@destengs.com."""
+from __future__ import annotations
+
+import email
+import email.utils
+import imaplib
+import smtplib
+import ssl
+from email.message import EmailMessage
+from email.policy import default as default_policy
+
+IMAP_HOST, IMAP_PORT = "mail.destengs.com", 993
+SMTP_HOST, SMTP_PORT = "mail.destengs.com", 587
+KEYWORD = "$ProjektChecked"
+
+
+class MailBox:
+    def __init__(self, user, password, conn=None):
+        self.conn = conn or imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT)
+        if conn is None:
+            self.conn.login(user, password)
+        self.conn.select("INBOX")
+        self._trash = None
+
+    def unchecked_uids(self):
+        typ, data = self.conn.uid("SEARCH", None, f"UNKEYWORD {KEYWORD}")
+        if typ != "OK" or not data or not data[0]:
+            return []
+        return [u.decode() for u in data[0].split()]
+
+    def fetch(self, uid):
+        typ, data = self.conn.uid("FETCH", uid, "(BODY.PEEK[])")
+        if typ != "OK" or not data or not data[0] or not isinstance(data[0], tuple):
+            raise RuntimeError(f"FETCH {uid} fehlgeschlagen: {typ}")
+        return email.message_from_bytes(data[0][1], policy=default_policy)
+
+    def flag_checked(self, uid):
+        self.conn.uid("STORE", uid, "+FLAGS", f"({KEYWORD})")
+
+    def trash_folder(self):
+        if self._trash:
+            return self._trash
+        typ, data = self.conn.list()
+        candidates = []
+        for raw in data or []:
+            line = raw.decode() if isinstance(raw, bytes) else str(raw)
+            name = line.rsplit(" ", 1)[-1].strip('"')
+            if "\\Trash" in line.split(")")[0]:
+                self._trash = name
+                return name
+            candidates.append(name)
+        for cand in ("Trash", "INBOX.Trash"):
+            if cand in candidates:
+                self._trash = cand
+                return cand
+        self._trash = "Trash"
+        return self._trash
+
+    def move_to_trash(self, uid):
+        folder = self.trash_folder()
+        typ, _ = self.conn.uid("MOVE", uid, folder)
+        if typ != "OK":
+            self.conn.uid("COPY", uid, folder)
+            self.conn.uid("STORE", uid, "+FLAGS", "(\\Deleted)")
+            self.conn.expunge()
+
+    def close(self):
+        try:
+            self.conn.logout()
+        except Exception:
+            pass
+
+
+def send_mail(user, password, to, subject, body):
+    msg = EmailMessage()
+    msg["From"], msg["To"], msg["Subject"] = user, to, subject
+    msg["Date"] = email.utils.formatdate(localtime=True)
+    msg["Message-ID"] = email.utils.make_msgid(domain="destengs.com")
+    msg.set_content(body)
+    with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=30) as server:
+        server.starttls(context=ssl.create_default_context())
+        server.login(user, password)
+        server.send_message(msg)

projekt-matching/projektmatch/mailparse.py

@@ -0,0 +1,83 @@
+"""E-mail parsing: own-mail detection, project splitting, canonical URLs."""
+from __future__ import annotations
+
+import re
+from urllib.parse import urlsplit, urlunsplit
+
+import requests
+from bs4 import BeautifulSoup
+
+OWN_PREFIXES = ("[Projekt-Match]", "[Projekt-Match-Fehler]")
+PROJECT_URL_RE = re.compile(
+    r"https?://(?:www\.)?freelancermap\.de/(?:nproj/|projekt/)[^\s\"'<>)\]]+")
+BROWSER_UA = ("Mozilla/5.0 (X11; Linux x86_64; rv:128.0) "
+              "Gecko/20100101 Firefox/128.0")
+
+
+def is_own_mail(subject) -> bool:
+    return bool(subject) and subject.strip().startswith(OWN_PREFIXES)
+
+
+def bodies(msg):
+    """Return (html, text) of the first text/html and text/plain parts."""
+    html = text = ""
+    for part in msg.walk():
+        if part.get_content_maintype() == "multipart":
+            continue
+        try:
+            payload = part.get_content()
+        except Exception:
+            continue
+        ctype = part.get_content_type()
+        if ctype == "text/html" and not html:
+            html = payload
+        elif ctype == "text/plain" and not text:
+            text = payload
+    return html, text
+
+
+def _url_key(url: str) -> str:
+    return urlsplit(url).path
+
+
+def split_projects(html: str, text: str) -> list:
+    """Extract project items {title, url} from a mail body. HTML preferred;
+    per project (URL path) the LONGEST anchor text wins (skips 'Zum Projekt'
+    buttons). Plain-text fallback: URL line + nearest preceding non-empty line."""
+    best = {}          # key -> {"title", "url"}
+    order = []
+    if html:
+        soup = BeautifulSoup(html, "html.parser")
+        for a in soup.find_all("a", href=PROJECT_URL_RE):
+            url = a["href"]
+            key = _url_key(url)
+            title = " ".join(a.get_text(" ", strip=True).split())
+            if key not in best:
+                best[key] = {"title": title, "url": url}
+                order.append(key)
+            elif len(title) > len(best[key]["title"]):
+                best[key]["title"] = title
+    if not best and text:
+        last_line = ""
+        for line in text.splitlines():
+            match = PROJECT_URL_RE.search(line)
+            stripped = line.strip()
+            if match:
+                key = _url_key(match.group(0))
+                if key not in best:
+                    best[key] = {"title": last_line or match.group(0),
+                                 "url": match.group(0)}
+                    order.append(key)
+            elif stripped:
+                last_line = stripped
+    return [best[k] for k in order if best[k]["title"]]
+
+
+def canonical_url(url: str, session=None, timeout=20) -> str:
+    """Follow redirects, strip query string and fragment."""
+    sess = session or requests.Session()
+    resp = sess.get(url, timeout=timeout, allow_redirects=True,
+                    headers={"User-Agent": BROWSER_UA})
+    resp.raise_for_status()
+    parts = urlsplit(resp.url)
+    return urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))

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

projekt-matching/projektmatch/rules.py

@@ -0,0 +1,106 @@
+"""Deterministic rules: Misc evaluation, match calculation, description markdown.
+
+Replicates SKILL.md Schritt 3b/4/5. No LLM here — the numbers written to the
+CRM must never be hallucinated.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import json
+import urllib.parse
+import urllib.request
+from math import asin, cos, radians, sin, sqrt
+
+SAUERLACH = (47.9721357, 11.6528398)
+NOMINATIM_UA = "projekt-matching/1.0 (Thomas.Langer@destengs.com)"
+
+YES, NO, UNKNOWN = "yes", "no", "unknown"
+SYMBOL = {YES: "✅", NO: "❌", UNKNOWN: "❔"}
+RATING_VALUE = {YES: 100, UNKNOWN: 25, NO: 0}
+KAT_ORDER = {"Must": 0, "Nice": 1, "Misc": 2}
+
+
+def geocode(place: str):
+    """Nominatim lookup -> (lat, lon) or None. Policy: max 1 request/s."""
+    url = "https://nominatim.openstreetmap.org/search?" + urllib.parse.urlencode(
+        {"q": place, "format": "json", "countrycodes": "de", "limit": 1})
+    req = urllib.request.Request(url, headers={"User-Agent": NOMINATIM_UA})
+    data = json.load(urllib.request.urlopen(req, timeout=20))
+    return (float(data[0]["lat"]), float(data[0]["lon"])) if data else None
+
+
+def distance_to_sauerlach_km(place: str, geocode_fn=geocode):
+    """Haversine distance; None on geocoding failure/ambiguity."""
+    try:
+        pos = geocode_fn(place)
+    except Exception:
+        return None
+    if pos is None:
+        return None
+    dlat = radians(pos[0] - SAUERLACH[0])
+    dlon = radians(pos[1] - SAUERLACH[1])
+    h = (sin(dlat / 2) ** 2
+         + cos(radians(SAUERLACH[0])) * cos(radians(pos[0])) * sin(dlon / 2) ** 2)
+    return 2 * 6371 * asin(sqrt(h))
+
+
+def eval_misc(item: dict, today: dt.date, geocode_fn=geocode) -> str:
+    kind = item.get("miscType") or "other"
+    if kind == "security":
+        return NO                          # K.-o.-Kriterium (SÜ dauert >= 3 Monate)
+    if kind == "start":
+        iso = item.get("startDate")
+        if not iso:
+            return UNKNOWN
+        try:
+            start = dt.date.fromisoformat(iso)
+        except ValueError:
+            return UNKNOWN
+        return YES if start <= today + dt.timedelta(days=56) else NO
+    if kind == "workload":
+        pct = item.get("workloadPercent")
+        if pct is None:
+            return UNKNOWN
+        return YES if 75 <= pct <= 100 else UNKNOWN
+    if kind == "duration":
+        return YES
+    if kind == "location":
+        if item.get("remotePercent") == 100:
+            return YES
+        place = item.get("onsiteLocation")
+        if not place:
+            return NO                      # Einsatzort/Remote-Anteil unklar
+        km = distance_to_sauerlach_km(place, geocode_fn)
+        if km is None:
+            return UNKNOWN
+        if km <= 50:
+            return YES
+        if km <= 60:
+            return UNKNOWN
+        return NO
+    return UNKNOWN
+
+
+def calc_match(rows: list, kat: str):
+    vals = [RATING_VALUE[r["rating"]] for r in rows if r["kat"] == kat]
+    if not vals:
+        return None
+    return int(sum(vals) / len(vals) + 0.5)    # half-up, not banker's rounding
+
+
+def fmt_match(value):
+    return "–" if value is None else f"{value} %"
+
+
+def order_rows(rows: list) -> list:
+    return sorted(rows, key=lambda r: KAT_ORDER[r["kat"]])   # stable sort
+
+
+def build_description(rows: list) -> str:
+    """rows already ordered Must -> Nice -> Misc (use order_rows)."""
+    must, nice = calc_match(rows, "Must"), calc_match(rows, "Nice")
+    lines = [f"Must-have-Match: {fmt_match(must)} · Nice-to-have-Match: {fmt_match(nice)}",
+             "", "| Nr. | Kat. | ❔ | Anforderung |", "|---|---|---|---|"]
+    for i, r in enumerate(rows, 1):
+        lines.append(f"| {i} | {r['kat']} | {SYMBOL[r['rating']]} | {r['text']} |")
+    return "\n".join(lines)

projekt-matching/projektmatch/stages.py

@@ -0,0 +1,175 @@
+"""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 team not in (back.get("teamsIds") or []):
+        problems.append("teamsIds")
+    if account_id and agency and back.get("cAccount1Id") != account_id:
+        problems.append("cAccount1Id")
+    if account_id and not agency and back.get("accountId") != account_id:
+        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):
+    """Send notification on created; ALWAYS post the Langfuse trace.
+
+    Called directly by the component (not via run_stage), so failed runs
+    still get their trace and an SMTP error cannot lose it."""
+    try:
+        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)
+    except Exception as exc:  # noqa: BLE001
+        ctx.update(status="failed", error=f"notify: {exc}")
+    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)

projekt-matching/pyproject.toml

@@ -0,0 +1,13 @@
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "projektmatch"
+version = "0.1.0"
+description = "Automated freelancermap project matching (Langflow backend)"
+dependencies = ["requests", "beautifulsoup4"]
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["projektmatch*"]

projekt-matching/pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+testpaths = tests

projekt-matching/tests/e2e/check_state.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+"""Report current system state: CRM hits for a token, INBOX/Trash mails."""
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+from projektmatch import mailer               # noqa: E402
+from projektmatch.espocrm import EspoClient   # noqa: E402
+
+token = sys.argv[1] if len(sys.argv) > 1 else ""
+espo = EspoClient(os.environ["PM_ESPO_BASE"], os.environ["PM_ESPO_API_KEY"])
+if token:
+    hits = espo.search("Opportunity", "contains", "cProjektlink", token,
+                       select="name,cProjektlink")
+    print(f"CRM hits for '{token}': {len(hits)}")
+    for h in hits:
+        print(f"  {h['id']}  {h['name']}")
+box = mailer.MailBox("chancen@destengs.com", os.environ["PM_IMAP_PASSWORD"])
+for folder in ("INBOX", box.trash_folder()):
+    box.conn.select(folder)
+    typ, data = box.conn.uid("SEARCH", None, "ALL")
+    uids = data[0].split() if data and data[0] else []
+    print(f"{folder}: {len(uids)} mails")
+    for uid in uids[-8:]:
+        typ, d = box.conn.uid("FETCH", uid.decode(),
+                              "(BODY.PEEK[HEADER.FIELDS (SUBJECT)])")
+        print("  ", d[0][1].decode(errors="replace").strip())
+box.close()

projekt-matching/tests/e2e/cleanup_crm.py

@@ -0,0 +1,19 @@
+#!/usr/bin/env python3
+"""Delete test opportunities whose cProjektlink contains a token."""
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+from projektmatch.espocrm import EspoClient  # noqa: E402
+
+TOKENS = sys.argv[1:] or [
+    "python-entwickler-fuer-eine-ki-anwendung",
+    "test-engineer-m-w-d-3020995",
+    "nproj/3020338", "nproj/3020995"]
+espo = EspoClient(os.environ["PM_ESPO_BASE"], os.environ["PM_ESPO_API_KEY"])
+for token in TOKENS:
+    for hit in espo.search("Opportunity", "contains", "cProjektlink", token,
+                           select="name,cProjektlink", max_size=100):
+        espo.delete("Opportunity", hit["id"])
+        print(f"deleted {hit['id']} {hit['name']}")
+print("cleanup done")

projekt-matching/tests/e2e/flag_inbox.py

@@ -0,0 +1,13 @@
+#!/usr/bin/env python3
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+from projektmatch import mailer  # noqa: E402
+
+box = mailer.MailBox("chancen@destengs.com", os.environ["PM_IMAP_PASSWORD"])
+uids = box.unchecked_uids()
+for uid in uids:
+    box.flag_checked(uid)
+print(f"flagged {len(uids)} mails as checked")
+box.close()

projekt-matching/tests/e2e/run_flow2.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+"""Run Flow 2 directly via the Langflow run API from the host."""
+import json
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+from projektmatch.config import Cfg          # noqa: E402
+from projektmatch.ingest import dispatch_flow2  # noqa: E402
+
+env = dict(line.split("=", 1) for line in open(
+    os.path.expanduser("~/.config/projekt-matching/env"))
+    if "=" in line.strip())
+cfg = Cfg(flow2_id=env["FLOW2_ID"].strip(),
+          langflow_api_key=env["LANGFLOW_API_KEY"].strip(),
+          langflow_base="http://127.0.0.1:8090")
+result = dispatch_flow2(cfg, {"canonical": sys.argv[1],
+                              "title": sys.argv[2] if len(sys.argv) > 2
+                              else "Test"})
+print(json.dumps(result, indent=2, ensure_ascii=False))

projekt-matching/tests/e2e/send_test_mail.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+"""Re-send an original example mail (found by subject) as a fresh test mail
+from chancen@ to chancen@. Fallback: synthetic single-project mail."""
+import os
+import smtplib
+import ssl
+import sys
+import time
+from email.message import EmailMessage
+from email.policy import default as default_policy
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+from projektmatch import mailer, mailparse    # noqa: E402
+
+USER = "chancen@destengs.com"
+PW = os.environ["PM_IMAP_PASSWORD"]
+SUBJECT = sys.argv[1]        # e.g. "Example 1 for Langflow process"
+FALLBACK_URL = sys.argv[2] if len(sys.argv) > 2 else ""
+
+box = mailer.MailBox(USER, PW)
+html = ""
+for folder in ("INBOX", box.trash_folder()):
+    box.conn.select(folder)
+    typ, data = box.conn.uid("SEARCH", None,
+                             f'SUBJECT "{SUBJECT}"')
+    uids = data[0].split() if data and data[0] else []
+    if uids:
+        msg = box.fetch(uids[-1].decode())
+        html, _ = mailparse.bodies(msg)
+        break
+box.close()
+
+out = EmailMessage(policy=default_policy)
+out["From"], out["To"] = USER, USER
+out["Subject"] = f"Test {SUBJECT} {int(time.time())}"
+if html:
+    out.set_content("siehe HTML")
+    out.add_alternative(html, subtype="html")
+    print(f"re-sending original '{SUBJECT}' body")
+else:
+    if not FALLBACK_URL:
+        sys.exit(f"original mail '{SUBJECT}' not found and no fallback URL")
+    out.set_content(f"Neues Projekt:\n\nTestprojekt\n{FALLBACK_URL}\n")
+    print("original not found -> synthetic fallback mail")
+with smtplib.SMTP(mailer.SMTP_HOST, mailer.SMTP_PORT, timeout=30) as server:
+    server.starttls(context=ssl.create_default_context())
+    server.login(USER, PW)
+    server.send_message(out)
+print(f"sent: {out['Subject']}")

projekt-matching/tests/test_espocrm.py

@@ -0,0 +1,70 @@
+from unittest import mock
+
+import pytest
+
+from projektmatch import espocrm
+
+
+def make_client(responses):
+    """Client whose session returns queued mock responses."""
+    client = espocrm.EspoClient("https://crm.example/api/v1", "k", retries=1,
+                                backoff=0)
+    client.session = mock.Mock()
+    client.session.request.side_effect = responses
+    return client
+
+
+def resp(status=200, body=None, reason=""):
+    r = mock.Mock()
+    r.status_code = status
+    r.json.return_value = body if body is not None else {}
+    r.text = "x" if body is not None else ""
+    r.headers = {"X-Status-Reason": reason}
+    return r
+
+
+def test_find_opportunity_by_link():
+    client = make_client([resp(body={"list": [{"id": "1", "name": "P"}]})])
+    assert client.find_opportunity_by_link("https://x")["id"] == "1"
+    params = client.session.request.call_args.kwargs["params"]
+    assert params["where[0][type]"] == "equals"
+    assert params["where[0][attribute]"] == "cProjektlink"
+
+
+def test_team_id_missing_raises():
+    client = make_client([resp(body={"list": []})])
+    with pytest.raises(espocrm.EspoError, match="DesTEngS"):
+        client.team_id("DesTEngS")
+
+
+def test_ensure_account_exact_match_and_create():
+    hits = {"list": [{"id": "a1", "name": "Aristo Group", "type": "Reseller"}]}
+    client = make_client([resp(body=hits)])
+    assert client.ensure_account("Aristo Group", "Reseller") == "a1"
+    client = make_client([resp(body={"list": []}), resp(body={"id": "a2"})])
+    assert client.ensure_account("Neue GmbH", "Customer") == "a2"
+    payload = client.session.request.call_args.kwargs["json"]
+    assert payload == {"name": "Neue GmbH", "type": "Customer"}
+
+
+def test_unique_opportunity_name_suffix():
+    hits = {"list": [{"name": "Projekt X"}, {"name": "Projekt X (2)"}]}
+    client = make_client([resp(body=hits)])
+    assert client.unique_opportunity_name("Projekt X") == "Projekt X (3)"
+
+
+def test_retry_on_500_then_success():
+    client = make_client([resp(status=500), resp(body={"list": []})])
+    assert client.search("Team", "equals", "name", "X") == []
+
+
+def test_400_raises_with_reason():
+    client = make_client([resp(status=400, body={}, reason="bad field")])
+    with pytest.raises(espocrm.EspoError, match="bad field"):
+        client.search("Team", "equals", "name", "X")
+
+
+def test_helpers():
+    assert espocrm.core_token("Aristo Group GmbH") == "Aristo"
+    assert espocrm.split_person("Max Muster Mann") == ("Max Muster", "Mann")
+    assert espocrm.split_person("Mann") == ("", "Mann")

projekt-matching/tests/test_ingest.py

@@ -0,0 +1,163 @@
+import json
+
+import pytest
+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_unknown_dispatch_status_normalized_to_failed(tmp_path):
+    mails = {"1": mail("Mail", PROJECT_HTML)}
+    summary, box, sent = run(tmp_path, mails,
+                             dispatch_result={"weird": True})
+    assert summary["failed"] == 1
+    box.move_to_trash.assert_called_once()
+
+
+def test_alert_send_failure_still_trashes(tmp_path):
+    mails = {"1": mail("Mail", PROJECT_HTML)}
+    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 = None
+    def boom_send(*a):
+        raise OSError("smtp down")
+    with mock.patch("projektmatch.ingest.mailparse.bodies",
+                    side_effect=lambda m: (m["html"], "")), \
+         mock.patch("projektmatch.ingest.mailer.send_mail", boom_send):
+        summary = ingest.run_ingest(
+            cfg(tmp_path), mailbox=box, espo=espo,
+            resolve=lambda url: "https://www.freelancermap.de/projekt/eins",
+            dispatch=lambda item: {"status": "failed", "error": "x"})
+    assert summary["alertFailed"] == 1
+    box.move_to_trash.assert_called_once_with("1")
+
+
+def test_unreadable_mail_flagged_pipeline_continues(tmp_path):
+    good = mail("Example 2 for Langflow process", PROJECT_HTML)
+    box = mock.Mock()
+    box.unchecked_uids.return_value = ["1", "2"]
+
+    def fetch_side_effect(uid):
+        if uid == "1":
+            raise RuntimeError("FETCH 1 fehlgeschlagen: NO")
+        return good
+    box.fetch.side_effect = fetch_side_effect
+    espo = mock.Mock()
+    espo.find_opportunity_by_link.return_value = None
+    with mock.patch("projektmatch.ingest.mailparse.bodies",
+                    side_effect=lambda m: (m["html"], "")), \
+         mock.patch("projektmatch.ingest.mailer.send_mail",
+                    lambda *a: None):
+        summary = ingest.run_ingest(
+            cfg(tmp_path), mailbox=box, espo=espo,
+            resolve=lambda url: "https://www.freelancermap.de/projekt/eins",
+            dispatch=lambda item: {"status": "created"})
+    assert summary["unreadable"] == 1
+    assert summary["created"] == 1
+    box.flag_checked.assert_called_once_with("1")
+    box.move_to_trash.assert_called_once_with("2")
+    box.close.assert_called_once()
+
+
+def test_close_called_even_if_unchecked_uids_raises(tmp_path):
+    box = mock.Mock()
+    box.unchecked_uids.side_effect = RuntimeError("boom")
+    espo = mock.Mock()
+    with pytest.raises(RuntimeError):
+        ingest.run_ingest(cfg(tmp_path), mailbox=box, espo=espo)
+    box.close.assert_called_once()
+
+
+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"

projekt-matching/tests/test_llm.py

@@ -0,0 +1,69 @@
+import json
+from unittest import mock
+
+import pytest
+
+from projektmatch import llm
+
+
+def fake_post(payloads):
+    """Return a mock for requests.post yielding chat completions."""
+    responses = []
+    for p in payloads:
+        r = mock.Mock()
+        r.status_code = 200
+        r.json.return_value = {"choices": [{"message": {"content": p}}]}
+        responses.append(r)
+    return mock.Mock(side_effect=responses)
+
+
+def test_chat_json_strips_think_and_parses():
+    content = "<think>lange Kette</think>{\"a\": 1}"
+    with mock.patch("projektmatch.llm.requests.post", fake_post([content])) as p:
+        out = llm.chat_json("http://v/v1", "m", [{"role": "user", "content": "x"}],
+                            {"type": "object"})
+    assert out == {"a": 1}
+    body = p.call_args.kwargs["json"]
+    assert "max_tokens" not in body
+    assert body["chat_template_kwargs"] == {"enable_thinking": False}
+    assert body["response_format"]["json_schema"]["schema"] == {"type": "object"}
+
+
+def test_chat_json_strips_markdown_fences():
+    content = "```json\n{\"a\": 2}\n```"
+    with mock.patch("projektmatch.llm.requests.post", fake_post([content])):
+        out = llm.chat_json("http://v/v1", "m", [], {"type": "object"})
+    assert out == {"a": 2}
+
+
+def test_chat_json_retries_on_empty_content():
+    good = json.dumps({"a": 1})
+    with mock.patch("projektmatch.llm.time.sleep", lambda s: None), \
+         mock.patch("projektmatch.llm.requests.post", fake_post(["", good])):
+        assert llm.chat_json("http://v/v1", "m", [], {"type": "object"}) == {"a": 1}
+
+
+def test_chat_json_raises_after_exhausted_retries():
+    with mock.patch("projektmatch.llm.time.sleep", lambda s: None), \
+         mock.patch("projektmatch.llm.requests.post", fake_post(["", "", ""])):
+        with pytest.raises(llm.LlmError, match="kein JSON"):
+            llm.chat_json("http://v/v1", "m", [], {"type": "object"})
+
+
+def test_match_cv_retries_on_missing_nr_then_raises():
+    reqs = [{"nr": 1, "text": "Python"}, {"nr": 2, "text": "K8s"}]
+    partial = json.dumps({"ratings": [
+        {"nr": 1, "rating": "yes", "reason": "ok"}]})
+    with mock.patch("projektmatch.llm.requests.post",
+                    fake_post([partial, partial])):
+        with pytest.raises(llm.LlmError, match="unvollständig"):
+            llm.match_cv("http://v/v1", "m", "CV", reqs)
+
+
+def test_match_cv_ok():
+    reqs = [{"nr": 1, "text": "Python"}]
+    full = json.dumps({"ratings": [
+        {"nr": 1, "rating": "unknown", "reason": "Teilevidenz"}]})
+    with mock.patch("projektmatch.llm.requests.post", fake_post([full])):
+        out = llm.match_cv("http://v/v1", "m", "CV", reqs)
+    assert out[0]["rating"] == "unknown"

projekt-matching/tests/test_mailer.py

@@ -0,0 +1,73 @@
+import email
+
+import pytest
+from unittest import mock
+
+from projektmatch import mailer
+
+
+def fake_conn():
+    conn = mock.Mock()
+    conn.uid.return_value = ("OK", [b"3 7"])
+    conn.list.return_value = ("OK", [
+        b'(\\HasNoChildren) "." "INBOX"',
+        b'(\\HasNoChildren \\Trash) "." "Trash"'])
+    return conn
+
+
+def test_unchecked_uids_search():
+    box = mailer.MailBox("u", "p", conn=fake_conn())
+    assert box.unchecked_uids() == ["3", "7"]
+    box.conn.uid.assert_called_with("SEARCH", None,
+                                    "UNKEYWORD $ProjektChecked")
+
+
+def test_flag_checked():
+    box = mailer.MailBox("u", "p", conn=fake_conn())
+    box.flag_checked("3")
+    box.conn.uid.assert_called_with("STORE", "3", "+FLAGS",
+                                    "($ProjektChecked)")
+
+
+def test_trash_folder_by_special_use():
+    box = mailer.MailBox("u", "p", conn=fake_conn())
+    assert box.trash_folder() == "Trash"
+
+
+def test_move_uses_move_then_fallback():
+    conn = fake_conn()
+    box = mailer.MailBox("u", "p", conn=conn)
+    conn.uid.return_value = ("OK", [b""])
+    box.move_to_trash("3")
+    assert conn.uid.call_args_list[-1].args[:2] == ("MOVE", "3")
+    conn.uid.side_effect = [("NO", [b""]), ("OK", [b""]), ("OK", [b""])]
+    box.move_to_trash("4")
+    assert conn.expunge.called
+
+
+def test_fetch_raises_on_no_response():
+    conn = fake_conn()
+    conn.uid.return_value = ("NO", [None])
+    box = mailer.MailBox("u", "p", conn=conn)
+    with pytest.raises(RuntimeError):
+        box.fetch("3")
+
+
+def test_fetch_raises_on_malformed_ok_response():
+    conn = fake_conn()
+    conn.uid.return_value = ("OK", [None])
+    box = mailer.MailBox("u", "p", conn=conn)
+    with pytest.raises(RuntimeError):
+        box.fetch("3")
+
+
+def test_send_mail_starttls():
+    with mock.patch("projektmatch.mailer.smtplib.SMTP") as smtp:
+        server = smtp.return_value.__enter__.return_value
+        mailer.send_mail("u@x", "pw", "to@x", "Subj", "Body")
+    server.starttls.assert_called_once()
+    server.login.assert_called_once_with("u@x", "pw")
+    msg = server.send_message.call_args.args[0]
+    assert msg["Subject"] == "Subj" and msg["To"] == "to@x"
+    assert msg["Date"]
+    assert msg["Message-ID"].endswith("@destengs.com>")

projekt-matching/tests/test_mailparse.py

@@ -0,0 +1,69 @@
+import email
+from email.policy import default as default_policy
+from unittest import mock
+
+from projektmatch import mailparse
+
+HTML = """
+<html><body>
+<h2><a href="https://www.freelancermap.de/nproj/3020338.html?utm_source=x&t=1">
+Python Entwickler für eine KI-Anwendung (m/w/d)</a></h2>
+<p>Beschreibung ...</p>
+<a href="https://www.freelancermap.de/nproj/3020338.html?utm_source=x&t=1">Zum Projekt</a>
+<h2><a href="https://www.freelancermap.de/nproj/3020995.html?agent=2">
+Test Engineer (m/w/d)</a></h2>
+<a href="https://www.freelancermap.de/other/page.html">Impressum</a>
+</body></html>
+"""
+
+TEXT = """Neue Projekte fuer Sie:
+
+Python Entwickler für eine KI-Anwendung (m/w/d)
+https://www.freelancermap.de/nproj/3020338.html?utm_source=x
+
+Test Engineer (m/w/d)
+https://www.freelancermap.de/nproj/3020995.html?agent=2
+"""
+
+
+def test_is_own_mail():
+    assert mailparse.is_own_mail("[Projekt-Match] Freelancermap — 86 % — X")
+    assert mailparse.is_own_mail("[Projekt-Match-Fehler] Freelancermap — 2 Projekt(e)")
+    assert not mailparse.is_own_mail("Example 1 for Langflow process")
+    assert not mailparse.is_own_mail(None)
+
+
+def test_split_projects_html_dedupes_and_prefers_long_title():
+    items = mailparse.split_projects(HTML, "")
+    assert [i["title"] for i in items] == [
+        "Python Entwickler für eine KI-Anwendung (m/w/d)", "Test Engineer (m/w/d)"]
+    assert "3020338" in items[0]["url"]
+
+
+def test_split_projects_text_fallback():
+    items = mailparse.split_projects("", TEXT)
+    assert len(items) == 2
+    assert items[0]["title"] == "Python Entwickler für eine KI-Anwendung (m/w/d)"
+
+
+def test_split_projects_none_for_normal_mail():
+    assert mailparse.split_projects("<p>Hallo</p>", "Hallo") == []
+
+
+def test_bodies_multipart():
+    msg = email.message.EmailMessage(policy=default_policy)
+    msg["Subject"] = "x"
+    msg.set_content("plain body")
+    msg.add_alternative("<p>html body</p>", subtype="html")
+    html, text = mailparse.bodies(msg)
+    assert "html body" in html and "plain body" in text
+
+
+def test_canonical_url_strips_query():
+    session = mock.Mock()
+    session.get.return_value = mock.Mock(
+        url="https://www.freelancermap.de/projekt/python-entwickler?ref=1",
+        raise_for_status=lambda: None)
+    out = mailparse.canonical_url("https://www.freelancermap.de/nproj/1.html?x=1",
+                                  session=session)
+    assert out == "https://www.freelancermap.de/projekt/python-entwickler"

projekt-matching/tests/test_rules.py

@@ -0,0 +1,96 @@
+import datetime as dt
+
+from projektmatch import rules
+
+TODAY = dt.date(2026, 7, 9)
+
+
+def geo_munich(place):
+    return (48.137, 11.575)          # ~40 km from Sauerlach
+
+
+def geo_hamburg(place):
+    return (53.551, 9.994)           # ~600 km
+
+
+def geo_fail(place):
+    raise OSError("network down")
+
+
+def misc(**kw):
+    base = {"miscType": "other", "startDate": None, "workloadPercent": None,
+            "remotePercent": None, "onsiteLocation": None}
+    base.update(kw)
+    return base
+
+
+def test_security_is_always_no():
+    assert rules.eval_misc(misc(miscType="security"), TODAY) == rules.NO
+
+
+def test_start_within_8_weeks_yes_after_no_missing_unknown():
+    assert rules.eval_misc(misc(miscType="start", startDate="2026-08-01"), TODAY) == rules.YES
+    assert rules.eval_misc(misc(miscType="start", startDate="2026-12-01"), TODAY) == rules.NO
+    assert rules.eval_misc(misc(miscType="start"), TODAY) == rules.UNKNOWN
+    assert rules.eval_misc(misc(miscType="start", startDate="asap"), TODAY) == rules.UNKNOWN
+
+
+def test_workload():
+    assert rules.eval_misc(misc(miscType="workload", workloadPercent=100), TODAY) == rules.YES
+    assert rules.eval_misc(misc(miscType="workload", workloadPercent=50), TODAY) == rules.UNKNOWN
+    assert rules.eval_misc(misc(miscType="workload"), TODAY) == rules.UNKNOWN
+
+
+def test_duration_always_yes():
+    assert rules.eval_misc(misc(miscType="duration"), TODAY) == rules.YES
+
+
+def test_location_rules():
+    full_remote = misc(miscType="location", remotePercent=100)
+    assert rules.eval_misc(full_remote, TODAY) == rules.YES
+    hybrid_near = misc(miscType="location", remotePercent=60, onsiteLocation="München")
+    assert rules.eval_misc(hybrid_near, TODAY, geocode_fn=geo_munich) == rules.YES
+    hybrid_far = misc(miscType="location", remotePercent=60, onsiteLocation="Hamburg")
+    assert rules.eval_misc(hybrid_far, TODAY, geocode_fn=geo_hamburg) == rules.NO
+    unclear = misc(miscType="location")
+    assert rules.eval_misc(unclear, TODAY) == rules.NO
+    geo_broken = misc(miscType="location", remotePercent=40, onsiteLocation="Xyzstadt")
+    assert rules.eval_misc(geo_broken, TODAY, geocode_fn=geo_fail) == rules.UNKNOWN
+
+
+def test_calc_match_and_rounding():
+    rows = [{"kat": "Must", "rating": "yes", "text": "a"},
+            {"kat": "Must", "rating": "yes", "text": "b"},
+            {"kat": "Must", "rating": "unknown", "text": "c"},
+            {"kat": "Nice", "rating": "no", "text": "d"},
+            {"kat": "Misc", "rating": "yes", "text": "e"}]
+    assert rules.calc_match(rows, "Must") == 75          # (100+100+25)/3
+    assert rules.calc_match(rows, "Nice") == 0
+    assert rules.calc_match([], "Nice") is None
+    # commercial rounding, half up (not banker's): (100+25)/2 = 62.5 -> 63
+    two = [{"kat": "Must", "rating": "yes", "text": "a"},
+           {"kat": "Must", "rating": "unknown", "text": "b"}]
+    assert rules.calc_match(two, "Must") == 63
+
+
+def test_build_description_exact_format():
+    rows = [{"kat": "Misc", "rating": "yes", "text": "Hybrid: 60 % remote"},
+            {"kat": "Must", "rating": "no", "text": "Mehrjährige LLM-Plattform-Erfahrung"},
+            {"kat": "Must", "rating": "yes", "text": "RAG-Systeme"},
+            {"kat": "Nice", "rating": "yes", "text": "vLLM"}]
+    text = rules.build_description(rules.order_rows(rows))
+    assert text == (
+        "Must-have-Match: 50 % · Nice-to-have-Match: 100 %\n"
+        "\n"
+        "| Nr. | Kat. | ❔ | Anforderung |\n"
+        "|---|---|---|---|\n"
+        "| 1 | Must | ❌ | Mehrjährige LLM-Plattform-Erfahrung |\n"
+        "| 2 | Must | ✅ | RAG-Systeme |\n"
+        "| 3 | Nice | ✅ | vLLM |\n"
+        "| 4 | Misc | ✅ | Hybrid: 60 % remote |")
+
+
+def test_empty_category_dash():
+    rows = [{"kat": "Must", "rating": "yes", "text": "a"}]
+    assert rules.build_description(rows).startswith(
+        "Must-have-Match: 100 % · Nice-to-have-Match: –")

projekt-matching/tests/test_smoke.py

@@ -0,0 +1,5 @@
+import projektmatch
+
+
+def test_package_imports():
+    assert projektmatch.__doc__

projekt-matching/tests/test_stages.py

@@ -0,0 +1,109 @@
+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
+    traced = []
+    with mock.patch("projektmatch.stages.mailer.send_mail",
+                    lambda *a: sent.append(a)), \
+         mock.patch("projektmatch.stages.pm_trace.post_trace",
+                    lambda c: traced.append(c["status"])):
+        stages.stage_notify({"status": "failed", "error": "x"}, CFG)
+    assert traced == ["failed"] and not sent