diff --git a/decnet/templates/_shared/auth-helper/auth-helper.c b/decnet/templates/_shared/auth-helper/auth-helper.c
index 4454497d..bee66102 100644
--- a/decnet/templates/_shared/auth-helper/auth-helper.c
+++ b/decnet/templates/_shared/auth-helper/auth-helper.c
@@ -13,12 +13,23 @@
  *     55555, MSGID `auth_attempt` (matches FTP's existing event type so
  *     the parser + dashboard pick it up with zero changes).
  *
- * Two password fields ride in the SD-block:
- *   password      RFC 5424-escaped ASCII-printable, '?' for non-printables.
- *                 FTP-compatible; consumed by existing dashboard rendering.
- *   password_b64  base64 of the exact PAM_AUTHTOK bytes. Lossless.
- *                 Preserves NUL/0xff/control bytes that the plain field
- *                 would silently drop — useful fingerprinting signal.
+ * SD-block carries the standardized credential shape (matches
+ * decnet/web/db/models/logs.py:Credential). Universal keys consumed
+ * directly by the ingester's native-shape branch:
+ *   principal         the human-meaningful identity the attacker sent
+ *                     (username for SSH/Telnet; would be a domain for
+ *                     SMTP, a DN for LDAP, etc.)
+ *   secret_printable  RFC 5424-escaped ASCII-printable, '?' for non-
+ *                     printables. Best-effort display form; may be
+ *                     lossy on non-UTF8 bytes.
+ *   secret_b64        base64 of the exact PAM_AUTHTOK bytes. Lossless.
+ *                     Preserves NUL/0xff/control bytes that the plain
+ *                     field would silently drop — useful fingerprinting
+ *                     signal that survives display sanitization.
+ *
+ * `username` rides alongside as a service-specific identity field for
+ * SSH/Telnet (mirrors `principal`); future emitters (SMTP, LDAP, …)
+ * drop `username` in favor of their service-native identity field.
  *
  * Fail-open: every error path silently exits 0. The PAM line is `optional`
  * so a malfunctioning helper must never break sshd auth.
@@ -150,13 +161,19 @@ pw_done:;
     b64_encode(pw_raw, pw_len, pw_b64, sizeof(pw_b64));
 
     /* Priority: facility=local0(16), severity=INFO(6) → <16*8+6> = <134>.
-     * Matches the syslog_bridge.py default exactly. */
+     * Matches the syslog_bridge.py default exactly.
+     *
+     * SD-block keys match the Credential storage model: principal +
+     * secret_printable + secret_b64 are the universal keys the ingester
+     * keys off; username is emitted alongside principal so existing
+     * dashboards that read SSH/Telnet `username=` keep working until
+     * the cred-reuse UI lands. */
     char line[LINE_BUF];
     int n = snprintf(line, sizeof(line),
         "<134>1 %s %s auth-helper - auth_attempt "
-        "[relay@55555 username=\"%s\" password=\"%s\" "
-        "password_b64=\"%s\" src_ip=\"%s\"]\n",
-        tsbuf, host, user_esc, pw_esc, pw_b64, rhost_esc);
+        "[relay@55555 username=\"%s\" principal=\"%s\" "
+        "secret_printable=\"%s\" secret_b64=\"%s\" src_ip=\"%s\"]\n",
+        tsbuf, host, user_esc, user_esc, pw_esc, pw_b64, rhost_esc);
     if (n <= 0 || (size_t)n >= sizeof(line)) return 0;
 
     /* /proc/1/fd/1 is the entrypoint's stdout — the fd Docker captures
diff --git a/decnet/templates/ssh/auth-helper/auth-helper.c b/decnet/templates/ssh/auth-helper/auth-helper.c
index 4454497d..bee66102 100644
--- a/decnet/templates/ssh/auth-helper/auth-helper.c
+++ b/decnet/templates/ssh/auth-helper/auth-helper.c
@@ -13,12 +13,23 @@
  *     55555, MSGID `auth_attempt` (matches FTP's existing event type so
  *     the parser + dashboard pick it up with zero changes).
  *
- * Two password fields ride in the SD-block:
- *   password      RFC 5424-escaped ASCII-printable, '?' for non-printables.
- *                 FTP-compatible; consumed by existing dashboard rendering.
- *   password_b64  base64 of the exact PAM_AUTHTOK bytes. Lossless.
- *                 Preserves NUL/0xff/control bytes that the plain field
- *                 would silently drop — useful fingerprinting signal.
+ * SD-block carries the standardized credential shape (matches
+ * decnet/web/db/models/logs.py:Credential). Universal keys consumed
+ * directly by the ingester's native-shape branch:
+ *   principal         the human-meaningful identity the attacker sent
+ *                     (username for SSH/Telnet; would be a domain for
+ *                     SMTP, a DN for LDAP, etc.)
+ *   secret_printable  RFC 5424-escaped ASCII-printable, '?' for non-
+ *                     printables. Best-effort display form; may be
+ *                     lossy on non-UTF8 bytes.
+ *   secret_b64        base64 of the exact PAM_AUTHTOK bytes. Lossless.
+ *                     Preserves NUL/0xff/control bytes that the plain
+ *                     field would silently drop — useful fingerprinting
+ *                     signal that survives display sanitization.
+ *
+ * `username` rides alongside as a service-specific identity field for
+ * SSH/Telnet (mirrors `principal`); future emitters (SMTP, LDAP, …)
+ * drop `username` in favor of their service-native identity field.
  *
  * Fail-open: every error path silently exits 0. The PAM line is `optional`
  * so a malfunctioning helper must never break sshd auth.
@@ -150,13 +161,19 @@ pw_done:;
     b64_encode(pw_raw, pw_len, pw_b64, sizeof(pw_b64));
 
     /* Priority: facility=local0(16), severity=INFO(6) → <16*8+6> = <134>.
-     * Matches the syslog_bridge.py default exactly. */
+     * Matches the syslog_bridge.py default exactly.
+     *
+     * SD-block keys match the Credential storage model: principal +
+     * secret_printable + secret_b64 are the universal keys the ingester
+     * keys off; username is emitted alongside principal so existing
+     * dashboards that read SSH/Telnet `username=` keep working until
+     * the cred-reuse UI lands. */
     char line[LINE_BUF];
     int n = snprintf(line, sizeof(line),
         "<134>1 %s %s auth-helper - auth_attempt "
-        "[relay@55555 username=\"%s\" password=\"%s\" "
-        "password_b64=\"%s\" src_ip=\"%s\"]\n",
-        tsbuf, host, user_esc, pw_esc, pw_b64, rhost_esc);
+        "[relay@55555 username=\"%s\" principal=\"%s\" "
+        "secret_printable=\"%s\" secret_b64=\"%s\" src_ip=\"%s\"]\n",
+        tsbuf, host, user_esc, user_esc, pw_esc, pw_b64, rhost_esc);
     if (n <= 0 || (size_t)n >= sizeof(line)) return 0;
 
     /* /proc/1/fd/1 is the entrypoint's stdout — the fd Docker captures
diff --git a/decnet/templates/telnet/auth-helper/auth-helper.c b/decnet/templates/telnet/auth-helper/auth-helper.c
index 4454497d..bee66102 100644
--- a/decnet/templates/telnet/auth-helper/auth-helper.c
+++ b/decnet/templates/telnet/auth-helper/auth-helper.c
@@ -13,12 +13,23 @@
  *     55555, MSGID `auth_attempt` (matches FTP's existing event type so
  *     the parser + dashboard pick it up with zero changes).
  *
- * Two password fields ride in the SD-block:
- *   password      RFC 5424-escaped ASCII-printable, '?' for non-printables.
- *                 FTP-compatible; consumed by existing dashboard rendering.
- *   password_b64  base64 of the exact PAM_AUTHTOK bytes. Lossless.
- *                 Preserves NUL/0xff/control bytes that the plain field
- *                 would silently drop — useful fingerprinting signal.
+ * SD-block carries the standardized credential shape (matches
+ * decnet/web/db/models/logs.py:Credential). Universal keys consumed
+ * directly by the ingester's native-shape branch:
+ *   principal         the human-meaningful identity the attacker sent
+ *                     (username for SSH/Telnet; would be a domain for
+ *                     SMTP, a DN for LDAP, etc.)
+ *   secret_printable  RFC 5424-escaped ASCII-printable, '?' for non-
+ *                     printables. Best-effort display form; may be
+ *                     lossy on non-UTF8 bytes.
+ *   secret_b64        base64 of the exact PAM_AUTHTOK bytes. Lossless.
+ *                     Preserves NUL/0xff/control bytes that the plain
+ *                     field would silently drop — useful fingerprinting
+ *                     signal that survives display sanitization.
+ *
+ * `username` rides alongside as a service-specific identity field for
+ * SSH/Telnet (mirrors `principal`); future emitters (SMTP, LDAP, …)
+ * drop `username` in favor of their service-native identity field.
  *
  * Fail-open: every error path silently exits 0. The PAM line is `optional`
  * so a malfunctioning helper must never break sshd auth.
@@ -150,13 +161,19 @@ pw_done:;
     b64_encode(pw_raw, pw_len, pw_b64, sizeof(pw_b64));
 
     /* Priority: facility=local0(16), severity=INFO(6) → <16*8+6> = <134>.
-     * Matches the syslog_bridge.py default exactly. */
+     * Matches the syslog_bridge.py default exactly.
+     *
+     * SD-block keys match the Credential storage model: principal +
+     * secret_printable + secret_b64 are the universal keys the ingester
+     * keys off; username is emitted alongside principal so existing
+     * dashboards that read SSH/Telnet `username=` keep working until
+     * the cred-reuse UI lands. */
     char line[LINE_BUF];
     int n = snprintf(line, sizeof(line),
         "<134>1 %s %s auth-helper - auth_attempt "
-        "[relay@55555 username=\"%s\" password=\"%s\" "
-        "password_b64=\"%s\" src_ip=\"%s\"]\n",
-        tsbuf, host, user_esc, pw_esc, pw_b64, rhost_esc);
+        "[relay@55555 username=\"%s\" principal=\"%s\" "
+        "secret_printable=\"%s\" secret_b64=\"%s\" src_ip=\"%s\"]\n",
+        tsbuf, host, user_esc, user_esc, pw_esc, pw_b64, rhost_esc);
     if (n <= 0 || (size_t)n >= sizeof(line)) return 0;
 
     /* /proc/1/fd/1 is the entrypoint's stdout — the fd Docker captures
diff --git a/decnet/web/db/models/__init__.py b/decnet/web/db/models/__init__.py
index 63d04a09..6de83d61 100644
--- a/decnet/web/db/models/__init__.py
+++ b/decnet/web/db/models/__init__.py
@@ -48,6 +48,8 @@ from .health import (
 from .logs import (
     Bounty,
     BountyResponse,
+    Credential,
+    CredentialsResponse,
     Log,
     LogsResponse,
     State,
@@ -167,6 +169,8 @@ __all__ = [
     # logs
     "Bounty",
     "BountyResponse",
+    "Credential",
+    "CredentialsResponse",
     "Log",
     "LogsResponse",
     "State",
diff --git a/decnet/web/db/models/logs.py b/decnet/web/db/models/logs.py
index 0b5a6d8d..275a0cec 100644
--- a/decnet/web/db/models/logs.py
+++ b/decnet/web/db/models/logs.py
@@ -1,9 +1,9 @@
-"""Log / Bounty / State tables + their list-response DTOs."""
+"""Log / Bounty / Credential / State tables + their list-response DTOs."""
 from datetime import datetime, timezone
 from typing import Any, List, Optional
 
 from pydantic import BaseModel
-from sqlalchemy import Column, Text
+from sqlalchemy import Column, Index, Text
 from sqlmodel import Field, SQLModel
 
 from ._base import _BIG_TEXT
@@ -40,6 +40,58 @@ class Bounty(SQLModel, table=True):
     payload: str = Field(sa_column=Column("payload", Text, nullable=False))
 
 
+class Credential(SQLModel, table=True):
+    """One observed credential attempt against a decky service.
+
+    Forward-compatible across every auth-bearing service in the fleet:
+    SSH user+pass, Telnet user+pass, SMTP domain+pass, LDAP dn+pass,
+    Redis password-only, etc. The two universal lossless representations
+    (``secret_b64`` + ``secret_sha256``) hoist to indexed columns so
+    cross-service reuse queries don't scan opaque JSON.
+
+    Per-service identity (the human-meaningful "who's authenticating")
+    lives in ``principal`` — username for SSH, domain for SMTP, dn for
+    LDAP. Nullable for principal-less mechanisms (Redis AUTH, bearer
+    tokens). Fully service-specific keys ride in ``fields`` JSON.
+
+    Dedup contract: same (attacker_uuid, decky, service, secret_sha256,
+    principal_or_empty) tuple → upsert, bumps ``attempt_count`` and
+    ``last_seen``. Different secret or different principal → new row.
+    """
+    __tablename__ = "credentials"
+    __table_args__ = (
+        Index("ix_credentials_secret_service", "secret_sha256", "service"),
+        Index("ix_credentials_principal_service", "principal", "service"),
+    )
+    id: Optional[int] = Field(default=None, primary_key=True)
+    # Keyed by attacker IP (not attackers.uuid) to match Bounty's pattern
+    # and avoid the chicken-and-egg of writing a credential row before
+    # the profiler has minted the Attacker. Index covers the join path
+    # cred_reuse → Attacker.ip.
+    attacker_ip: str = Field(index=True)
+    decky_name: str = Field(index=True)
+    service: str = Field(index=True)
+    principal: Optional[str] = Field(default=None, index=True, max_length=256)
+    # Universal lossless secret representations.
+    secret_sha256: str = Field(index=True, max_length=64)
+    secret_b64: Optional[str] = Field(default=None, max_length=2048)
+    # Best-effort printable form — non-printable bytes collapsed to '?'
+    # by either auth-helper.c (SSH/Telnet) or the ingester's legacy
+    # adapter (FTP/POP3/IMAP/SMTP). May be lossy on non-UTF8.
+    secret_printable: Optional[str] = Field(default=None, max_length=512)
+    outcome: Optional[str] = Field(default=None, max_length=16)  # success|failure|observed
+    fields: str = Field(
+        sa_column=Column("fields", _BIG_TEXT, nullable=False, default="{}")
+    )
+    first_seen: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc), index=True
+    )
+    last_seen: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc), index=True
+    )
+    attempt_count: int = Field(default=1)
+
+
 class State(SQLModel, table=True):
     __tablename__ = "state"
     key: str = Field(primary_key=True)
@@ -62,6 +114,13 @@ class BountyResponse(BaseModel):
     data: List[dict[str, Any]]
 
 
+class CredentialsResponse(BaseModel):
+    total: int
+    limit: int
+    offset: int
+    data: List[dict[str, Any]]
+
+
 class StatsResponse(BaseModel):
     total_logs: int
     unique_attackers: int
diff --git a/decnet/web/db/repository.py b/decnet/web/db/repository.py
index e37295ba..22d0ea02 100644
--- a/decnet/web/db/repository.py
+++ b/decnet/web/db/repository.py
@@ -110,6 +110,55 @@ class BaseRepository(ABC):
         """Retrieve the total count of bounties, optionally filtered."""
         pass
 
+    # ---- credentials ---------------------------------------------------
+
+    @abstractmethod
+    async def upsert_credential(self, data: dict[str, Any]) -> int:
+        """Insert or upsert a credential attempt; returns the row id.
+
+        Dedup tuple: (attacker_ip, decky_name, service, secret_sha256,
+        principal_or_None). On dedup match, ``attempt_count`` is bumped
+        and ``last_seen`` updated; the originally-seen ``first_seen``
+        and ``fields`` JSON are preserved.
+        """
+        pass
+
+    @abstractmethod
+    async def get_credentials(
+        self,
+        limit: int = 50,
+        offset: int = 0,
+        search: Optional[str] = None,
+        service: Optional[str] = None,
+        attacker_ip: Optional[str] = None,
+    ) -> list[dict[str, Any]]:
+        """Paginated credential rows, with optional filters."""
+        pass
+
+    @abstractmethod
+    async def get_total_credentials(
+        self,
+        search: Optional[str] = None,
+        service: Optional[str] = None,
+        attacker_ip: Optional[str] = None,
+    ) -> int:
+        """Total credential count under the same filters as get_credentials."""
+        pass
+
+    @abstractmethod
+    async def get_credentials_for_attacker(
+        self, attacker_ip: str
+    ) -> list[dict[str, Any]]:
+        """Every credential row from the given attacker IP."""
+        pass
+
+    @abstractmethod
+    async def get_credential_reuse(
+        self, secret_sha256: str
+    ) -> list[dict[str, Any]]:
+        """Every (attacker, decky, service, principal) row sharing this secret hash."""
+        pass
+
     @abstractmethod
     async def get_state(self, key: str) -> Optional[dict[str, Any]]:
         """Retrieve a specific state entry by key."""
diff --git a/decnet/web/db/sqlmodel_repo.py b/decnet/web/db/sqlmodel_repo.py
index 358d5722..dd3855dd 100644
--- a/decnet/web/db/sqlmodel_repo.py
+++ b/decnet/web/db/sqlmodel_repo.py
@@ -31,6 +31,7 @@ from decnet.web.db.models import (
     User,
     Log,
     Bounty,
+    Credential,
     State,
     Attacker,
     AttackerBehavior,
@@ -448,6 +449,9 @@ class SQLModelRepository(BaseRepository):
         async with self._session() as session:
             logs_deleted = (await session.execute(text("DELETE FROM logs"))).rowcount
             bounties_deleted = (await session.execute(text("DELETE FROM bounty"))).rowcount
+            credentials_deleted = (
+                await session.execute(text("DELETE FROM credentials"))
+            ).rowcount
             # attacker_behavior has FK → attackers.uuid; delete children first.
             await session.execute(text("DELETE FROM attacker_behavior"))
             attackers_deleted = (await session.execute(text("DELETE FROM attackers"))).rowcount
@@ -455,6 +459,7 @@ class SQLModelRepository(BaseRepository):
         return {
             "logs": logs_deleted,
             "bounties": bounties_deleted,
+            "credentials": credentials_deleted,
             "attackers": attackers_deleted,
         }
 
@@ -535,6 +540,169 @@ class SQLModelRepository(BaseRepository):
             result = await session.execute(statement)
             return result.scalar() or 0
 
+    # ─── credentials ──────────────────────────────────────────────────────
+
+    async def upsert_credential(self, data: dict[str, Any]) -> int:
+        """Upsert a credential attempt; returns the row id.
+
+        Dedup tuple: (attacker_ip, decky_name, service, secret_sha256,
+        principal_or_None). On match, ``attempt_count`` += 1 and
+        ``last_seen`` advances; ``first_seen`` and ``fields`` are
+        preserved from the original sighting.
+        """
+        payload = dict(data)
+        if "fields" in payload and isinstance(payload["fields"], dict):
+            # ensure_ascii=True keeps utf8mb4 columns safe even when
+            # service-specific keys carry non-ASCII bytes.
+            payload["fields"] = json.dumps(payload["fields"], ensure_ascii=True)
+
+        principal = payload.get("principal")
+        async with self._session() as session:
+            stmt = select(Credential).where(
+                Credential.attacker_ip == payload["attacker_ip"],
+                Credential.decky_name == payload["decky_name"],
+                Credential.service == payload["service"],
+                Credential.secret_sha256 == payload["secret_sha256"],
+                # NULL == NULL is False under SQL — branch the predicate.
+                (Credential.principal == principal) if principal is not None
+                else Credential.principal.is_(None),
+            )
+            existing = (await session.execute(stmt)).scalar_one_or_none()
+            now = datetime.now(timezone.utc)
+            if existing is not None:
+                existing.attempt_count = (existing.attempt_count or 1) + 1
+                existing.last_seen = now
+                if payload.get("outcome") is not None:
+                    existing.outcome = payload["outcome"]
+                session.add(existing)
+                await session.commit()
+                return existing.id  # type: ignore[return-value]
+            row = Credential(
+                attacker_ip=payload["attacker_ip"],
+                decky_name=payload["decky_name"],
+                service=payload["service"],
+                principal=principal,
+                secret_sha256=payload["secret_sha256"],
+                secret_b64=payload.get("secret_b64"),
+                secret_printable=payload.get("secret_printable"),
+                outcome=payload.get("outcome"),
+                fields=payload.get("fields", "{}"),
+                first_seen=now,
+                last_seen=now,
+                attempt_count=1,
+            )
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return row.id  # type: ignore[return-value]
+
+    def _apply_credential_filters(
+        self,
+        statement: SelectOfScalar,
+        search: Optional[str],
+        service: Optional[str],
+        attacker_ip: Optional[str],
+    ) -> SelectOfScalar:
+        if service:
+            statement = statement.where(Credential.service == service)
+        if attacker_ip:
+            statement = statement.where(Credential.attacker_ip == attacker_ip)
+        if search:
+            lk = f"%{search}%"
+            statement = statement.where(
+                or_(
+                    Credential.decky_name.like(lk),
+                    Credential.service.like(lk),
+                    Credential.principal.like(lk),
+                    Credential.secret_printable.like(lk),
+                )
+            )
+        return statement
+
+    async def get_credentials(
+        self,
+        limit: int = 50,
+        offset: int = 0,
+        search: Optional[str] = None,
+        service: Optional[str] = None,
+        attacker_ip: Optional[str] = None,
+    ) -> List[dict[str, Any]]:
+        statement = (
+            select(Credential)
+            .order_by(desc(Credential.last_seen))
+            .offset(offset)
+            .limit(limit)
+        )
+        statement = self._apply_credential_filters(
+            statement, search, service, attacker_ip
+        )
+        async with self._session() as session:
+            result = await session.execute(statement)
+            out: List[dict[str, Any]] = []
+            for item in result.scalars().all():
+                d = item.model_dump(mode="json")
+                try:
+                    d["fields"] = json.loads(d["fields"])
+                except (json.JSONDecodeError, TypeError):
+                    pass
+                out.append(d)
+            return out
+
+    async def get_total_credentials(
+        self,
+        search: Optional[str] = None,
+        service: Optional[str] = None,
+        attacker_ip: Optional[str] = None,
+    ) -> int:
+        statement = select(func.count()).select_from(Credential)
+        statement = self._apply_credential_filters(
+            statement, search, service, attacker_ip
+        )
+        async with self._session() as session:
+            result = await session.execute(statement)
+            return result.scalar() or 0
+
+    async def get_credentials_for_attacker(
+        self, attacker_ip: str
+    ) -> List[dict[str, Any]]:
+        async with self._session() as session:
+            result = await session.execute(
+                select(Credential)
+                .where(Credential.attacker_ip == attacker_ip)
+                .order_by(desc(Credential.last_seen))
+            )
+            out: List[dict[str, Any]] = []
+            for item in result.scalars().all():
+                d = item.model_dump(mode="json")
+                try:
+                    d["fields"] = json.loads(d["fields"])
+                except (json.JSONDecodeError, TypeError):
+                    pass
+                out.append(d)
+            return out
+
+    async def get_credential_reuse(
+        self, secret_sha256: str
+    ) -> List[dict[str, Any]]:
+        """Every (attacker_ip, decky, service, principal) row sharing this
+        secret hash. Indexed lookup via ix_credentials_secret_service.
+        """
+        async with self._session() as session:
+            result = await session.execute(
+                select(Credential)
+                .where(Credential.secret_sha256 == secret_sha256)
+                .order_by(desc(Credential.last_seen))
+            )
+            out: List[dict[str, Any]] = []
+            for item in result.scalars().all():
+                d = item.model_dump(mode="json")
+                try:
+                    d["fields"] = json.loads(d["fields"])
+                except (json.JSONDecodeError, TypeError):
+                    pass
+                out.append(d)
+            return out
+
     async def get_state(self, key: str) -> Optional[dict[str, Any]]:
         async with self._session() as session:
             statement = select(State).where(State.key == key)
diff --git a/decnet/web/ingester.py b/decnet/web/ingester.py
index 02fb68a4..cc4de7f3 100644
--- a/decnet/web/ingester.py
+++ b/decnet/web/ingester.py
@@ -1,5 +1,7 @@
 import asyncio
+import base64
 import contextlib
+import hashlib
 import ipaddress
 import os
 import json
@@ -195,6 +197,128 @@ async def _flush_batch(
     return _new_position
 
 
+# RFC 5424-ish SD-PARAM-VALUE sanitization, mirrored from auth-helper.c.
+# Bytes outside [0x20, 0x7f) collapse to '?', matching the C escape rule.
+# The hash is always computed over the *original* bytes so reuse queries
+# survive any sanitization on the printable form.
+_SECRET_B64_RE = re.compile(r"^[A-Za-z0-9+/]*={0,2}$")
+_SECRET_PRINTABLE_MAX = 512  # mirrors Credential.secret_printable max_length
+_PRINCIPAL_MAX = 256
+_SECRET_B64_MAX = 2048
+
+
+def _printable_filter(s: str) -> str:
+    """Replace bytes outside [0x20, 0x7f) with '?', matching auth-helper.c.
+
+    Operates on the str's UTF-8 encoded bytes so we don't accidentally
+    let a `\\u202e` Unicode override slip through display layers.
+    """
+    out: list[int] = []
+    for b in s.encode("utf-8", errors="replace"):
+        out.append(b if 0x20 <= b < 0x7f else ord("?"))
+    return bytes(out).decode("ascii")
+
+
+def _truncate_with_warn(s: Optional[str], cap: int, label: str) -> Optional[str]:
+    if s is None:
+        return None
+    if len(s) <= cap:
+        return s
+    logger.warning("ingester: %s truncated %d → %d chars", label, len(s), cap)
+    return s[:cap]
+
+
+async def _ingest_credential_native(
+    repo: BaseRepository,
+    log_data: dict[str, Any],
+    fields: dict[str, Any],
+) -> None:
+    """Native-shape credential: SD-block already carries secret_b64.
+
+    Validates the b64, computes sha256 over the decoded bytes, hands off
+    to the repo upsert. Drops the row on validation failure (the
+    underlying Log row still lands).
+    """
+    b64 = fields.get("secret_b64")
+    if not isinstance(b64, str) or not _SECRET_B64_RE.match(b64):
+        logger.warning(
+            "ingester: dropping credential — invalid secret_b64 from %s/%s",
+            log_data.get("decky"), log_data.get("service"),
+        )
+        return
+    try:
+        raw = base64.b64decode(b64, validate=True)
+    except (ValueError, TypeError):
+        logger.warning(
+            "ingester: dropping credential — secret_b64 decode failed from %s/%s",
+            log_data.get("decky"), log_data.get("service"),
+        )
+        return
+
+    sha256_hex = hashlib.sha256(raw).hexdigest()
+    principal = fields.get("principal") or fields.get("username")
+    secret_printable = fields.get("secret_printable")
+
+    await repo.upsert_credential({
+        "attacker_ip": log_data.get("attacker_ip"),
+        "decky_name": log_data.get("decky"),
+        "service": log_data.get("service"),
+        "principal": _truncate_with_warn(principal, _PRINCIPAL_MAX, "principal"),
+        "secret_sha256": sha256_hex,
+        "secret_b64": _truncate_with_warn(b64, _SECRET_B64_MAX, "secret_b64"),
+        "secret_printable": _truncate_with_warn(
+            secret_printable, _SECRET_PRINTABLE_MAX, "secret_printable"
+        ),
+        "outcome": fields.get("outcome"),
+        "fields": fields,  # repo handles json.dumps with ensure_ascii=True
+    })
+
+
+async def _ingest_credential_legacy(
+    repo: BaseRepository,
+    log_data: dict[str, Any],
+    fields: dict[str, Any],
+) -> None:
+    """Legacy-shape credential: SD-block has username + password.
+
+    Synthesizes secret_b64 (from utf8-encoded password bytes), the
+    sha256 hash (over those same bytes — lossless before any printable
+    sanitization), and a printable-filtered secret_printable. FTP /
+    POP3 / IMAP / SMTP go through this branch until DEBT-039 lands.
+    """
+    user = fields.get("username")
+    pw = fields.get("password")
+    if not isinstance(pw, str):
+        return
+
+    raw = pw.encode("utf-8", errors="replace")
+    sha256_hex = hashlib.sha256(raw).hexdigest()
+    b64 = base64.b64encode(raw).decode("ascii")
+    printable = _printable_filter(pw)
+
+    # Synthesize the universal keys into a copy of fields so the JSON
+    # blob carries the standardized shape too — lets downstream readers
+    # treat every credential row identically regardless of emitter.
+    synthesized_fields = dict(fields)
+    synthesized_fields.setdefault("principal", user)
+    synthesized_fields.setdefault("secret_printable", printable)
+    synthesized_fields.setdefault("secret_b64", b64)
+
+    await repo.upsert_credential({
+        "attacker_ip": log_data.get("attacker_ip"),
+        "decky_name": log_data.get("decky"),
+        "service": log_data.get("service"),
+        "principal": _truncate_with_warn(user, _PRINCIPAL_MAX, "principal"),
+        "secret_sha256": sha256_hex,
+        "secret_b64": _truncate_with_warn(b64, _SECRET_B64_MAX, "secret_b64"),
+        "secret_printable": _truncate_with_warn(
+            printable, _SECRET_PRINTABLE_MAX, "secret_printable"
+        ),
+        "outcome": fields.get("outcome"),
+        "fields": synthesized_fields,
+    })
+
+
 @_traced("ingester.extract_bounty")
 async def _extract_bounty(repo: BaseRepository, log_data: dict[str, Any]) -> None:
     """Detect and extract valuable artifacts (bounties) from log entries."""
@@ -202,21 +326,21 @@ async def _extract_bounty(repo: BaseRepository, log_data: dict[str, Any]) -> Non
     if not isinstance(_fields, dict):
         return
 
-    # 1. Credentials (User/Pass)
-    _user = _fields.get("username")
-    _pass = _fields.get("password")
-
-    if _user and _pass:
-        await repo.add_bounty({
-            "decky": log_data.get("decky"),
-            "service": log_data.get("service"),
-            "attacker_ip": log_data.get("attacker_ip"),
-            "bounty_type": "credential",
-            "payload": {
-                "username": _user,
-                "password": _pass
-            }
-        })
+    # 1. Credentials — fork on emitter shape.
+    #
+    # New shape (SSH/Telnet auth-helper, future emitters): SD-block
+    # carries `secret_b64` directly. Universal across services.
+    #
+    # Legacy shape (FTP/POP3/IMAP/SMTP today): SD-block has `username`
+    # + `password`. Adapter synthesizes `secret_b64` + `secret_sha256`
+    # on the fly so those services land in the same Credential table
+    # without requiring a per-template emitter rewrite. Tracked as
+    # DEBT-039 — eventually those services emit the new shape natively
+    # and this branch dies.
+    if "secret_b64" in _fields:
+        await _ingest_credential_native(repo, log_data, _fields)
+    elif _fields.get("username") and _fields.get("password"):
+        await _ingest_credential_legacy(repo, log_data, _fields)
 
     # 2. HTTP User-Agent fingerprint
     _h_raw = _fields.get("headers")
diff --git a/development/DEBT.md b/development/DEBT.md
index e2f5144e..c1396cd0 100644
--- a/development/DEBT.md
+++ b/development/DEBT.md
@@ -1,6 +1,6 @@
 # DECNET — Technical Debt Register
 
-> Last updated: 2026-04-25 — DEBT-038 opened (SSH PAM cred-capture limitations).
+> Last updated: 2026-04-25 — DEBT-039 opened (legacy cred emitters), Credential storage model landed.
 > Severity: 🔴 Critical · 🟠 High · 🟡 Medium · 🟢 Low
 
 ---
@@ -382,8 +382,26 @@ The SSH cred-capture path that closes the "real OpenSSH doesn't log attempted pa
 
 5. **Telnet had the same gap — closed in commit `f1026b4`.** Telnet's busybox-telnetd → `/bin/login` PAM stack didn't log attempted passwords either; the `auth-helper` binary is service-agnostic and was extended into `/etc/pam.d/login` via the same one-line PAM hook. The canonical source moved to `decnet/templates/_shared/auth-helper/auth-helper.c` and is synced into both ssh/ and telnet/ build contexts via `_sync_auth_helper_sources()` (mirrors the existing sessrec sync). Limitations 1–4 above apply equally to the telnet hook.
 
+6. **Standardized SD shape (DEBT-039 follow-up).** The auth-helper SD-block now emits the universal `principal` + `secret_printable` + `secret_b64` keys consumed directly by the ingester's native-shape branch and stored as hoisted columns on the new `Credential` table. `username` rides alongside as a service-specific identity field for SSH/Telnet. Future emitters drop `username` in favor of their service-native identity (`domain` for SMTP, `dn` for LDAP, …).
+
 **Status:** Open — document-only ticket tracking the architectural trade-offs of the v1 implementation. None of these are blocking; they're the things to know if the helper ever needs upgrading.
 
+### DEBT-039 — Migrate FTP/POP3/IMAP/SMTP emitters to standardized credential shape
+**Files:** `decnet/templates/ftp/server.py`, `decnet/templates/pop3/server.py`, `decnet/templates/imap/server.py`, `decnet/templates/smtp/server.py`, `decnet/web/ingester.py` (legacy adapter at `_ingest_credential_legacy`).
+
+The new `Credential` storage model (commit landing alongside this entry) writes one universal shape: `principal` + `secret_sha256` + `secret_b64` + `secret_printable`. SSH and Telnet auth-helper emit those keys natively. The four legacy services — FTP, POP3, IMAP, SMTP — still emit the old `username=` + `password=` shape, and the ingester carries a one-shot adapter (`_ingest_credential_legacy`) that synthesizes the universal keys on the fly.
+
+The adapter works correctly but couples ingester logic to an emitter shape we'd rather see go away. Per-service migration:
+
+1. **FTP** (`templates/ftp/server.py:103`) — change `_log("auth_attempt", username=..., password=...)` to also emit `principal`, `secret_printable`, `secret_b64`. Remove the legacy adapter dependency for `service="ftp"` once verified.
+2. **POP3** (`templates/pop3/server.py`) — same pattern.
+3. **IMAP** (`templates/imap/server.py`) — same pattern.
+4. **SMTP** (`templates/smtp/server.py`) — opportunity to use the new `domain` field as the principal (rather than `username` for an MTA), since SMTP AUTH PLAIN/LOGIN's authentication identity is conceptually a domain user, not a system user.
+
+Once all four migrate, delete `_ingest_credential_legacy` from `decnet/web/ingester.py` and drop the legacy branch from `_extract_bounty`.
+
+**Status:** Open — the legacy adapter is a temporary bridge. No deadline; close one service at a time as their templates are touched for unrelated reasons.
+
 ### DEBT-032 — Prober can't detect fingerprint rotation without mutation
 **Files:** `decnet/prober/worker.py` (~lines 235, 286, 334, 392), `decnet/web/db/models.py` (new `decky_service_fingerprints` table).
 
@@ -464,6 +482,7 @@ The prober already computes JARM (`worker.py:286`), HASSH (`worker.py:334`), and
 | DEBT-036 | 🟡 Medium | Correlation / Keystroke dynamics | open |
 | DEBT-037 | 🟡 Medium | Integration / Webhooks | open (tracks MVP follow-ups) |
 | DEBT-038 | 🟡 Medium | Honeypot / SSH cred capture | open (document-only) |
+| DEBT-039 | 🟡 Medium | Honeypot / Cred emitters | open |
 
-**Remaining open:** DEBT-011 (Alembic), DEBT-023 (image pinning), DEBT-026 (modular mailboxes), DEBT-027 (Dynamic bait store), DEBT-028 (deploy endpoint tests), DEBT-032 (fingerprint rotation detection), DEBT-033 (transcript shard rotation), DEBT-035 (artifacts uid/gid alignment), DEBT-036 (session-profile ingester), DEBT-037 (webhook delivery hardening), DEBT-038 (SSH PAM cred-capture limitations — document-only).
+**Remaining open:** DEBT-011 (Alembic), DEBT-023 (image pinning), DEBT-026 (modular mailboxes), DEBT-027 (Dynamic bait store), DEBT-028 (deploy endpoint tests), DEBT-032 (fingerprint rotation detection), DEBT-033 (transcript shard rotation), DEBT-035 (artifacts uid/gid alignment), DEBT-036 (session-profile ingester), DEBT-037 (webhook delivery hardening), DEBT-038 (SSH PAM cred-capture limitations — document-only), DEBT-039 (legacy cred emitters → standardized shape).
 **Estimated remaining effort:** ~24 hours. DEBT-030 Phase B (optimistic staged-buffer editor) is a follow-up, not debt.
diff --git a/tests/db/test_base_repo.py b/tests/db/test_base_repo.py
index 01f5fa1e..52121d68 100644
--- a/tests/db/test_base_repo.py
+++ b/tests/db/test_base_repo.py
@@ -19,6 +19,11 @@ class DummyRepo(BaseRepository):
     async def add_bounty(self, d): await super().add_bounty(d)
     async def get_bounties(self, **kw): await super().get_bounties(**kw)
     async def get_total_bounties(self, **kw): await super().get_total_bounties(**kw)
+    async def upsert_credential(self, d): await super().upsert_credential(d); return 0
+    async def get_credentials(self, **kw): await super().get_credentials(**kw)
+    async def get_total_credentials(self, **kw): await super().get_total_credentials(**kw)
+    async def get_credentials_for_attacker(self, ip): await super().get_credentials_for_attacker(ip)
+    async def get_credential_reuse(self, h): await super().get_credential_reuse(h)
     async def get_state(self, k): await super().get_state(k)
     async def set_state(self, k, v): await super().set_state(k, v)
     async def get_max_log_id(self): await super().get_max_log_id()
@@ -64,6 +69,11 @@ async def test_base_repo_coverage():
     await dr.add_bounty({})
     await dr.get_bounties()
     await dr.get_total_bounties()
+    await dr.upsert_credential({})
+    await dr.get_credentials()
+    await dr.get_total_credentials()
+    await dr.get_credentials_for_attacker("1.2.3.4")
+    await dr.get_credential_reuse("abc")
     await dr.get_state("k")
     await dr.set_state("k", "v")
     await dr.get_max_log_id()
diff --git a/tests/db/test_credentials.py b/tests/db/test_credentials.py
new file mode 100644
index 00000000..9e132b80
--- /dev/null
+++ b/tests/db/test_credentials.py
@@ -0,0 +1,142 @@
+"""Credential model + repo tests — upsert, dedup, cross-service reuse."""
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+import pytest
+
+from decnet.web.db.factory import get_repository
+
+
+@pytest.fixture
+async def repo(tmp_path: Path):
+    r = get_repository(db_path=str(tmp_path / "creds.db"))
+    await r.initialize()
+    return r
+
+
+def _sha256(s: str) -> str:
+    return hashlib.sha256(s.encode("utf-8")).hexdigest()
+
+
+@pytest.mark.anyio
+async def test_upsert_inserts_then_dedups(repo) -> None:
+    """Same dedup tuple twice → one row, attempt_count=2."""
+    payload = {
+        "attacker_ip": "10.0.0.5",
+        "decky_name": "decky-01",
+        "service": "ssh",
+        "principal": "root",
+        "secret_sha256": _sha256("hunter2"),
+        "secret_b64": "aHVudGVyMg==",
+        "secret_printable": "hunter2",
+        "fields": {"user": "root"},
+    }
+    rid_a = await repo.upsert_credential(payload)
+    rid_b = await repo.upsert_credential(payload)
+    assert rid_a == rid_b
+    rows = await repo.get_credentials()
+    assert len(rows) == 1
+    assert rows[0]["attempt_count"] == 2
+    assert rows[0]["fields"] == {"user": "root"}  # preserved
+
+
+@pytest.mark.anyio
+async def test_different_principal_creates_new_row(repo) -> None:
+    base = {
+        "attacker_ip": "10.0.0.5",
+        "decky_name": "decky-01",
+        "service": "ssh",
+        "secret_sha256": _sha256("hunter2"),
+        "secret_b64": "aHVudGVyMg==",
+        "secret_printable": "hunter2",
+        "fields": {},
+    }
+    await repo.upsert_credential({**base, "principal": "root"})
+    await repo.upsert_credential({**base, "principal": "admin"})
+    rows = await repo.get_credentials()
+    assert len(rows) == 2
+
+
+@pytest.mark.anyio
+async def test_null_principal_dedups_independently(repo) -> None:
+    """principal=None and principal='root' are different keys."""
+    base = {
+        "attacker_ip": "10.0.0.5",
+        "decky_name": "decky-01",
+        "service": "ssh",
+        "secret_sha256": _sha256("hunter2"),
+        "secret_b64": "aHVudGVyMg==",
+        "secret_printable": "hunter2",
+        "fields": {},
+    }
+    await repo.upsert_credential({**base, "principal": None})
+    await repo.upsert_credential({**base, "principal": None})  # dedupes
+    await repo.upsert_credential({**base, "principal": "root"})
+    rows = await repo.get_credentials()
+    assert len(rows) == 2
+    null_row = next(r for r in rows if r["principal"] is None)
+    assert null_row["attempt_count"] == 2
+
+
+@pytest.mark.anyio
+async def test_cross_service_reuse_query(repo) -> None:
+    """Same secret across SSH + FTP + SMTP → reuse query returns all three."""
+    secret = "hunter2"
+    sha = _sha256(secret)
+    services = [
+        ("ssh",  "decky-01", "root"),
+        ("ftp",  "decky-02", "anonymous"),
+        ("smtp", "decky-03", "acme.com"),
+    ]
+    for svc, decky, principal in services:
+        await repo.upsert_credential({
+            "attacker_ip": "10.0.0.5",
+            "decky_name": decky,
+            "service": svc,
+            "principal": principal,
+            "secret_sha256": sha,
+            "secret_b64": "aHVudGVyMg==",
+            "secret_printable": secret,
+            "fields": {},
+        })
+    reuse = await repo.get_credential_reuse(sha)
+    assert {r["service"] for r in reuse} == {"ssh", "ftp", "smtp"}
+
+
+@pytest.mark.anyio
+async def test_get_credentials_for_attacker(repo) -> None:
+    base = {
+        "decky_name": "decky-01",
+        "service": "ssh",
+        "principal": "root",
+        "secret_sha256": _sha256("hunter2"),
+        "secret_b64": "aHVudGVyMg==",
+        "secret_printable": "hunter2",
+        "fields": {},
+    }
+    await repo.upsert_credential({**base, "attacker_ip": "10.0.0.5"})
+    await repo.upsert_credential({**base, "attacker_ip": "10.0.0.6"})
+    rows = await repo.get_credentials_for_attacker("10.0.0.5")
+    assert len(rows) == 1
+    assert rows[0]["attacker_ip"] == "10.0.0.5"
+
+
+@pytest.mark.anyio
+async def test_filters(repo) -> None:
+    base_secret = _sha256("a")
+    await repo.upsert_credential({
+        "attacker_ip": "10.0.0.5", "decky_name": "decky-01", "service": "ssh",
+        "principal": "root", "secret_sha256": base_secret,
+        "secret_printable": "a", "fields": {},
+    })
+    await repo.upsert_credential({
+        "attacker_ip": "10.0.0.5", "decky_name": "decky-01", "service": "ftp",
+        "principal": "root", "secret_sha256": base_secret,
+        "secret_printable": "a", "fields": {},
+    })
+    rows = await repo.get_credentials(service="ssh")
+    assert len(rows) == 1 and rows[0]["service"] == "ssh"
+    assert await repo.get_total_credentials(service="ssh") == 1
+    assert await repo.get_total_credentials() == 2
diff --git a/tests/web/test_ingester.py b/tests/web/test_ingester.py
index aaa0458c..0ad82028 100644
--- a/tests/web/test_ingester.py
+++ b/tests/web/test_ingester.py
@@ -17,46 +17,117 @@ import pytest
 
 class TestExtractBounty:
     @pytest.mark.asyncio
-    async def test_credential_extraction(self):
+    async def test_credential_legacy_adapter(self):
+        """FTP/POP3/IMAP/SMTP shape (username + password) → upsert_credential."""
         from decnet.web.ingester import _extract_bounty
         mock_repo = MagicMock()
-        mock_repo.add_bounty = AsyncMock()
+        mock_repo.upsert_credential = AsyncMock()
         log_data: dict = {
             "decky": "decky-01",
-            "service": "ssh",
+            "service": "ftp",
             "attacker_ip": "10.0.0.5",
             "fields": {"username": "admin", "password": "hunter2"},
         }
         await _extract_bounty(mock_repo, log_data)
-        mock_repo.add_bounty.assert_awaited_once()
-        bounty = mock_repo.add_bounty.call_args[0][0]
-        assert bounty["bounty_type"] == "credential"
-        assert bounty["payload"]["username"] == "admin"
-        assert bounty["payload"]["password"] == "hunter2"
+        mock_repo.upsert_credential.assert_awaited_once()
+        cred = mock_repo.upsert_credential.call_args[0][0]
+        assert cred["service"] == "ftp"
+        assert cred["principal"] == "admin"
+        assert cred["secret_printable"] == "hunter2"
+        # b64 + sha256 computed over the original utf-8 bytes.
+        import base64, hashlib
+        assert cred["secret_b64"] == base64.b64encode(b"hunter2").decode()
+        assert cred["secret_sha256"] == hashlib.sha256(b"hunter2").hexdigest()
+
+    @pytest.mark.asyncio
+    async def test_credential_native_shape(self):
+        """SSH/Telnet auth-helper shape (secret_b64) → upsert_credential."""
+        from decnet.web.ingester import _extract_bounty
+        import base64, hashlib
+        mock_repo = MagicMock()
+        mock_repo.upsert_credential = AsyncMock()
+        log_data: dict = {
+            "decky": "decky-01",
+            "service": "ssh",
+            "attacker_ip": "10.0.0.5",
+            "fields": {
+                "username": "root",
+                "principal": "root",
+                "secret_printable": "hunter2",
+                "secret_b64": base64.b64encode(b"hunter2").decode(),
+            },
+        }
+        await _extract_bounty(mock_repo, log_data)
+        mock_repo.upsert_credential.assert_awaited_once()
+        cred = mock_repo.upsert_credential.call_args[0][0]
+        assert cred["service"] == "ssh"
+        assert cred["principal"] == "root"
+        assert cred["secret_sha256"] == hashlib.sha256(b"hunter2").hexdigest()
+
+    @pytest.mark.asyncio
+    async def test_credential_native_invalid_b64_dropped(self):
+        """Malformed secret_b64 → row dropped with a warning, no upsert."""
+        from decnet.web.ingester import _extract_bounty
+        mock_repo = MagicMock()
+        mock_repo.upsert_credential = AsyncMock()
+        log_data: dict = {
+            "decky": "decky-01",
+            "service": "ssh",
+            "attacker_ip": "10.0.0.5",
+            "fields": {"secret_b64": "not!base64!!"},
+        }
+        await _extract_bounty(mock_repo, log_data)
+        mock_repo.upsert_credential.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_credential_legacy_sanitizes_nonprintable(self):
+        """Non-printable bytes in legacy password collapse to '?' in
+        secret_printable; b64 + sha256 reflect the ORIGINAL bytes."""
+        from decnet.web.ingester import _extract_bounty
+        import base64, hashlib
+        mock_repo = MagicMock()
+        mock_repo.upsert_credential = AsyncMock()
+        # ANSI escape + NUL byte in the password.
+        bad_pw = "\x1b[31mbad\x00trail"
+        log_data: dict = {
+            "decky": "decky-01",
+            "service": "ftp",
+            "attacker_ip": "10.0.0.5",
+            "fields": {"username": "user", "password": bad_pw},
+        }
+        await _extract_bounty(mock_repo, log_data)
+        cred = mock_repo.upsert_credential.call_args[0][0]
+        # No 0x1b, no NUL — collapsed to '?'.
+        assert "\x1b" not in cred["secret_printable"]
+        assert "\x00" not in cred["secret_printable"]
+        # Original bytes survive in b64 + sha256.
+        raw = bad_pw.encode("utf-8")
+        assert base64.b64decode(cred["secret_b64"]) == raw
+        assert cred["secret_sha256"] == hashlib.sha256(raw).hexdigest()
 
     @pytest.mark.asyncio
     async def test_no_fields_skips(self):
         from decnet.web.ingester import _extract_bounty
         mock_repo = MagicMock()
-        mock_repo.add_bounty = AsyncMock()
+        mock_repo.upsert_credential = AsyncMock()
         await _extract_bounty(mock_repo, {"decky": "x"})
-        mock_repo.add_bounty.assert_not_awaited()
+        mock_repo.upsert_credential.assert_not_awaited()
 
     @pytest.mark.asyncio
     async def test_fields_not_dict_skips(self):
         from decnet.web.ingester import _extract_bounty
         mock_repo = MagicMock()
-        mock_repo.add_bounty = AsyncMock()
+        mock_repo.upsert_credential = AsyncMock()
         await _extract_bounty(mock_repo, {"fields": "not-a-dict"})
-        mock_repo.add_bounty.assert_not_awaited()
+        mock_repo.upsert_credential.assert_not_awaited()
 
     @pytest.mark.asyncio
     async def test_missing_password_skips(self):
         from decnet.web.ingester import _extract_bounty
         mock_repo = MagicMock()
-        mock_repo.add_bounty = AsyncMock()
+        mock_repo.upsert_credential = AsyncMock()
         await _extract_bounty(mock_repo, {"fields": {"username": "admin"}})
-        mock_repo.add_bounty.assert_not_awaited()
+        mock_repo.upsert_credential.assert_not_awaited()
 
     @pytest.mark.asyncio
     async def test_missing_username_skips(self):