PAYG: hash-lineage detection primitives (modular extractor / store / detector) (#6464)

## What this is

Three orthogonal interfaces — each with one production impl — for
detecting whether an incoming tool call should join an existing process
via content-hash lineage. Groundwork for PR-I7a: nothing in this PR
calls the detector yet; the ingress/egress filter that wires it into
every controller lands separately.

Built to be modular along three axes. Swapping any of them should not
require changes elsewhere:

| Axis | Interface | V1 impl | Plausible future impl |
|---|---|---|---|
| Hash algorithm | `LineageSignatureExtractor` |
`ByteHashSignatureExtractor` (SHA-256) | `PdfMetadataSignatureExtractor`
(PDF `/ID`, content-stream hash) |
| Storage backend | `JobLineageStore` | `JpaJobLineageStore` |
`RedisJobLineageStore` |
| Matching policy | `HashLineageDetector` | `DefaultHashLineageDetector`
| strategy-driven variant (any-match-joins for multi-input — lives in
JobService) |

## Interfaces

### `LineageSignatureExtractor` — what counts as a fingerprint

```java
public interface LineageSignatureExtractor {
    Set<LineageSignature> extract(Path file) throws IOException;
    String name();
}
```

File-based (not stream-based) so a future PDF-aware extractor can open
the same file via jpdfium / PDFBox and pull `/ID[0]` or a content-stream
hash. Multiple extractors compose at the detector layer — Spring
auto-wires all `LineageSignatureExtractor` beans, the detector unions
their results.

Production impl: **`ByteHashSignatureExtractor`** — SHA-256 over the
file via 64 KiB-buffered `DigestInputStream`. Hardware-accelerated by
the JVM (Intel SHA-NI, ARM SHA).

### `JobLineageStore` — where signatures live

```java
public interface JobLineageStore {
    void record(UUID jobId, Set<LineageSignature> signatures, ArtifactKind kind);
    Optional<LineageMatch> findOpenJobForSignatures(Long userId, Set<LineageSignature> candidates, Duration window);
    int pruneOlderThan(Instant cutoff);
}
```

Knows nothing about storage technology. Production impl
**`JpaJobLineageStore`** runs a single joined query against
`job_artifact_hash` ⋈ `processing_job` — status + window filtering
happen at the database. The query is bounded by `Limit.of(1)` on the hot
path so a job set sharing a popular signature doesn't materialise
unwanted rows. A future `RedisJobLineageStore` (or write-through hybrid)
is a drop-in.

### `HashLineageDetector` — the high-level API

```java
public interface HashLineageDetector {
    Optional<LineageMatch> detect(Long userId, Path inputFile) throws IOException;
    void record(UUID jobId, Path file, ArtifactKind kind) throws IOException;
}
```

**`DefaultHashLineageDetector`** delegates extraction to every
registered `LineageSignatureExtractor`, storage to the configured
`JobLineageStore`, and reads `payg.lineage.workflow-window` (default
`PT5M`) from config. When a single extractor throws (e.g. a future
PDF-aware extractor against a malformed PDF), the other extractors still
contribute — failures don't block the byte-hash from landing.

## Profile gating

All three `@Component` beans (`JpaJobLineageStore`,
`ByteHashSignatureExtractor`, `DefaultHashLineageDetector`) are
`@Profile("saas")` — consistent with every other `:saas` bean. Without
this guard the JPA store would fail to wire against its profile-gated
repository in non-saas profiles that pull `:saas` onto the classpath.

## Tests

Run entirely in-memory; no database required.

- **`LineageSignatureTest`** — storage-key encoding round-trips, rejects
malformed `"type:value"` keys.
- **`ByteHashSignatureExtractorTest`** — identical bytes → identical
sigs; empty file hashes to the well-known SHA-256-of-empty constant; 10
MiB file streams without OOM.
- **`DefaultHashLineageDetectorTest`** — same-user / within-window /
status=OPEN filtering, multi-signature matching (one extractor sees
`pdf-id` and matches even when bytes differ), most-recent-job-wins,
record+detect round-trip, extractor-throwing-doesn't-break-others.

**`InMemoryJobLineageStore`** (in test sources) implements the same
`JobLineageStore` interface as the JPA impl, plus a `registerJob` hook
for tests to model job state. Same contract — proves the abstraction is
portable. When the Redis impl lands it gets the same contract tests.

## What's not in this PR (deliberate)

- The tool ingress/egress filter that wires the detector into every
controller — separate, focused review.
- `JobChargeService.openProcess()` — uses the detector, part of the
charging machinery, separate PR.
- Prune scheduler that calls `pruneOlderThan` — small follow-up
alongside the `shedlock` foundational table.
- PDF-aware extractor (`PdfMetadataSignatureExtractor`) — to be added
when we measure how often byte-hash-only misses real workflows.
- Multi-input "any-match-joins" lineage policy — that's a `JobService`
decision (PR-I7), not a primitive.

## Self-review pass applied

An independent code-review on this PR caught:

- **HIGH:** Missing `@Profile("saas")` on the three `@Component` beans →
fixed.
- **MEDIUM:** `pruneOlderThan` missing `@Transactional` (its
`@Modifying` query would have thrown
`InvalidDataAccessApiUsageException`) → fixed.
- **MEDIUM:** `findOpenJobsForSignatures` fetching the whole match set
just to `get(0)` → now takes `Limit`, JPA store passes `Limit.of(1)` on
the hot path.
- **LOW:** `InMemoryJobLineageStore` used both `synchronized` methods
and `ConcurrentHashMap` → dropped the redundant `ConcurrentHashMap`.

Deferred: project-wide UTC unification (`LocalDateTime.now()`
system-zone is the established convention; flipping one file mid-stack
caused a real test failure — proper fix needs its own audit).

## Rollback

Straight `git revert`. No callers yet; deleting these classes wouldn't
break anything.

---

## Checklist

- [x] Tests pass: `ENABLE_SAAS=true ./gradlew :saas:test`
- [x] No new warnings
- [x] Self-review performed (HIGH + MEDIUM findings addressed)

app/saas/src/main/java/stirling/software/saas/payg/lineage/ByteHashSignatureExtractor.java

@@ -0,0 +1,61 @@
+package stirling.software.saas.payg.lineage;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.DigestInputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.HexFormat;
+import java.util.Set;
+
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+/**
+ * SHA-256 of the file's bytes. The simplest universally-applicable signature — works for every
+ * content type, doesn't parse, doesn't allocate proportional to file size (fixed 64 KiB read
+ * buffer), hardware-accelerated by the JVM on modern hardware (Intel SHA-NI, ARM SHA extensions).
+ *
+ * <p>Always returns exactly one {@link LineageSignature} of type {@code "sha256"}. A future {@code
+ * PdfMetadataSignatureExtractor} would be a separate bean and add its own signature type — composed
+ * at the detector layer, no interaction needed here.
+ */
+@Component
+@Profile("saas")
+public class ByteHashSignatureExtractor implements LineageSignatureExtractor {
+
+    private static final String ALGORITHM = "SHA-256";
+    private static final String SIGNATURE_TYPE = "sha256";
+    private static final int BUFFER_SIZE = 64 * 1024;
+
+    @Override
+    public Set<LineageSignature> extract(Path file) throws IOException {
+        MessageDigest digest = newDigest();
+        try (InputStream raw = Files.newInputStream(file);
+                DigestInputStream in = new DigestInputStream(raw, digest)) {
+            byte[] buf = new byte[BUFFER_SIZE];
+            // Drain through the digest stream; we only care about side effects on the digest.
+            while (in.read(buf) != -1) {
+                // no-op
+            }
+        }
+        String hex = HexFormat.of().formatHex(digest.digest());
+        return Set.of(new LineageSignature(SIGNATURE_TYPE, hex));
+    }
+
+    @Override
+    public String name() {
+        return SIGNATURE_TYPE;
+    }
+
+    private static MessageDigest newDigest() {
+        try {
+            return MessageDigest.getInstance(ALGORITHM);
+        } catch (NoSuchAlgorithmException e) {
+            // SHA-256 is mandated by every JDK; unreachable in practice.
+            throw new IllegalStateException(ALGORITHM + " unavailable — JDK is misconfigured", e);
+        }
+    }
+}

app/saas/src/main/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetector.java

@@ -0,0 +1,111 @@
+package stirling.software.saas.payg.lineage;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.saas.payg.model.ArtifactKind;
+
+/**
+ * Default detector. Composes signatures from every registered {@link LineageSignatureExtractor}
+ * (currently just SHA-256 byte hash; future PDF-aware extractors are drop-in additions) and
+ * delegates lookup + persistence to a {@link JobLineageStore}.
+ */
+@Slf4j
+@Component
+@Profile("saas")
+public class DefaultHashLineageDetector implements HashLineageDetector {
+
+    private final List<LineageSignatureExtractor> extractors;
+    private final JobLineageStore store;
+    private final Duration workflowWindow;
+
+    public DefaultHashLineageDetector(
+            List<LineageSignatureExtractor> extractors,
+            JobLineageStore store,
+            @Value("${payg.lineage.workflow-window:PT5M}") Duration workflowWindow) {
+        Objects.requireNonNull(extractors, "extractors");
+        Objects.requireNonNull(store, "store");
+        Objects.requireNonNull(workflowWindow, "workflowWindow");
+        if (extractors.isEmpty()) {
+            throw new IllegalStateException(
+                    "DefaultHashLineageDetector requires at least one LineageSignatureExtractor"
+                            + " bean — none registered.");
+        }
+        if (workflowWindow.isNegative() || workflowWindow.isZero()) {
+            // A non-positive window would mean "since = now + |window|", so the > comparison
+            // only matches jobs in the future — i.e. nothing matches, silently. Fail loud
+            // instead.
+            throw new IllegalArgumentException(
+                    "payg.lineage.workflow-window must be positive, got " + workflowWindow);
+        }
+        this.extractors = List.copyOf(extractors);
+        this.store = store;
+        this.workflowWindow = workflowWindow;
+    }
+
+    @Override
+    public Optional<LineageMatch> detect(Long userId, Path inputFile) throws IOException {
+        Objects.requireNonNull(userId, "userId");
+        Objects.requireNonNull(inputFile, "inputFile");
+
+        Set<LineageSignature> signatures = extractAll(inputFile);
+        if (signatures.isEmpty()) {
+            // No extractor produced anything for this content. Treat as no-match.
+            log.debug("No signatures extracted from {}; lineage check returns empty.", inputFile);
+            return Optional.empty();
+        }
+
+        return store.findOpenJobForSignatures(userId, signatures, workflowWindow);
+    }
+
+    @Override
+    public void record(UUID jobId, Path file, ArtifactKind kind) throws IOException {
+        Objects.requireNonNull(jobId, "jobId");
+        Objects.requireNonNull(file, "file");
+        Objects.requireNonNull(kind, "kind");
+
+        Set<LineageSignature> signatures = extractAll(file);
+        if (signatures.isEmpty()) {
+            log.debug(
+                    "No signatures extracted from {} for job {} ({}); nothing recorded.",
+                    file,
+                    jobId,
+                    kind);
+            return;
+        }
+        store.record(jobId, signatures, kind);
+    }
+
+    private Set<LineageSignature> extractAll(Path file) {
+        Set<LineageSignature> union = new HashSet<>();
+        for (LineageSignatureExtractor extractor : extractors) {
+            try {
+                union.addAll(extractor.extract(file));
+            } catch (IOException e) {
+                // A single extractor failing on file IO / format parse (e.g. PDF-aware extractor
+                // on a malformed PDF) must not block other extractors from contributing. Log and
+                // continue. RuntimeExceptions deliberately propagate — they signal bugs we want
+                // to surface, not "expected" extractor-doesn't-fit-this-content failures.
+                log.debug(
+                        "Extractor '{}' failed on {} ({}); continuing with other extractors.",
+                        extractor.name(),
+                        file,
+                        e.getMessage());
+            }
+        }
+        return union;
+    }
+}

app/saas/src/main/java/stirling/software/saas/payg/lineage/HashLineageDetector.java

@@ -0,0 +1,36 @@
+package stirling.software.saas.payg.lineage;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Optional;
+import java.util.UUID;
+
+import stirling.software.saas.payg.model.ArtifactKind;
+
+/**
+ * High-level lineage API. Decides whether a tool call should join an existing open process by
+ * comparing the incoming file's signatures against signatures previously recorded for that user's
+ * open processes.
+ *
+ * <p>Two operations:
+ *
+ * <ul>
+ *   <li>{@link #detect} — pre-execution. Caller asks "for this user about to run a tool on this
+ *       file, is there an open process to join?"
+ *   <li>{@link #record} — post-execution (for outputs) or post-charge (for inputs). Caller records
+ *       the file's signatures against the job so subsequent tool calls can lineage-match on it.
+ * </ul>
+ *
+ * <p>Both methods take a {@link Path}; the caller is expected to have already materialised the
+ * upload / response body to a managed temp file (via {@code TempFileManager}). The detector
+ * delegates signature extraction to one or more {@link LineageSignatureExtractor}s and storage to a
+ * {@link JobLineageStore}, both of which are swappable.
+ */
+public interface HashLineageDetector {
+
+    /** Looks up an open process for this user whose recorded signatures match the input file. */
+    Optional<LineageMatch> detect(Long userId, Path inputFile) throws IOException;
+
+    /** Records the file's signatures against the given job as either INPUT or OUTPUT. */
+    void record(UUID jobId, Path file, ArtifactKind kind) throws IOException;
+}

app/saas/src/main/java/stirling/software/saas/payg/lineage/JobLineageStore.java

@@ -0,0 +1,39 @@
+package stirling.software.saas.payg.lineage;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+import stirling.software.saas.payg.model.ArtifactKind;
+
+/**
+ * Persistence boundary for lineage data. Two operations: record signatures against a job, and look
+ * up the open job (if any) that previously recorded a matching signature for the same user.
+ *
+ * <p>This interface deliberately knows nothing about the storage technology — the production impl
+ * is JPA-backed against {@code job_artifact_hash}, and a future Redis-backed impl will be a
+ * straight swap. The detector and any callers only see this interface.
+ *
+ * <p>Signature values are stored using {@link LineageSignature#asStorageKey()}, so multiple
+ * signature types (SHA-256, PDF {@code /ID}, etc.) coexist on the same table without a separate
+ * column.
+ */
+public interface JobLineageStore {
+
+    /** Records every signature in {@code signatures} against the given job + artifact kind. */
+    void record(UUID jobId, Set<LineageSignature> signatures, ArtifactKind kind);
+
+    /**
+     * Finds the most-recently-active open job (owned by {@code userId}, whose {@code last_step_at}
+     * is within {@code workflowWindow}) whose recorded artifacts include any of the supplied
+     * candidate signatures. Returns the first match; ordering across multiple matches is by job
+     * activity recency.
+     */
+    Optional<LineageMatch> findOpenJobForSignatures(
+            Long userId, Set<LineageSignature> candidates, Duration workflowWindow);
+
+    /** Deletes records created before {@code cutoff}. Returns the number of rows removed. */
+    int pruneOlderThan(Instant cutoff);
+}

app/saas/src/main/java/stirling/software/saas/payg/lineage/JpaJobLineageStore.java

@@ -0,0 +1,88 @@
+package stirling.software.saas.payg.lineage;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+import org.springframework.context.annotation.Profile;
+import org.springframework.data.domain.Limit;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import lombok.RequiredArgsConstructor;
+
+import stirling.software.saas.payg.job.JobArtifactHash;
+import stirling.software.saas.payg.job.JobArtifactHash.JobArtifactHashId;
+import stirling.software.saas.payg.model.ArtifactKind;
+import stirling.software.saas.payg.model.JobStatus;
+import stirling.software.saas.payg.repository.JobArtifactHashRepository;
+
+/**
+ * JPA-backed {@link JobLineageStore} against the {@code job_artifact_hash} table. The lookup runs
+ * as a single joined query against {@code processing_job} so status + window filtering happens at
+ * the database, not in-process.
+ *
+ * <p>Signatures are persisted using {@link LineageSignature#asStorageKey()} ({@code "type:value"})
+ * so multiple signature types coexist on the same column without a schema change.
+ */
+@Component
+@Profile("saas")
+@RequiredArgsConstructor
+public class JpaJobLineageStore implements JobLineageStore {
+
+    private final JobArtifactHashRepository hashRepository;
+
+    @Override
+    @Transactional
+    public void record(UUID jobId, Set<LineageSignature> signatures, ArtifactKind kind) {
+        Objects.requireNonNull(jobId, "jobId");
+        Objects.requireNonNull(signatures, "signatures");
+        Objects.requireNonNull(kind, "kind");
+        if (signatures.isEmpty()) {
+            return;
+        }
+        // saveAll + @Transactional → one transaction, all-or-nothing. Without this, a multi-
+        // signature record() could leave partial state if a save mid-way fails.
+        List<JobArtifactHash> rows = new ArrayList<>(signatures.size());
+        for (LineageSignature signature : signatures) {
+            JobArtifactHash row = new JobArtifactHash();
+            row.setId(new JobArtifactHashId(jobId, signature.asStorageKey(), kind));
+            rows.add(row);
+        }
+        hashRepository.saveAll(rows);
+    }
+
+    @Override
+    public Optional<LineageMatch> findOpenJobForSignatures(
+            Long userId, Set<LineageSignature> candidates, Duration workflowWindow) {
+        Objects.requireNonNull(userId, "userId");
+        Objects.requireNonNull(candidates, "candidates");
+        Objects.requireNonNull(workflowWindow, "workflowWindow");
+        if (candidates.isEmpty()) {
+            return Optional.empty();
+        }
+
+        List<String> storageKeys = candidates.stream().map(LineageSignature::asStorageKey).toList();
+        LocalDateTime since = LocalDateTime.now().minus(workflowWindow);
+
+        List<LineageMatch> matches =
+                hashRepository.findOpenJobsForSignatures(
+                        userId, JobStatus.OPEN, since, storageKeys, Limit.of(1));
+        return matches.isEmpty() ? Optional.empty() : Optional.of(matches.get(0));
+    }
+
+    @Override
+    @Transactional
+    public int pruneOlderThan(Instant cutoff) {
+        Objects.requireNonNull(cutoff, "cutoff");
+        return hashRepository.deleteOlderThan(
+                LocalDateTime.ofInstant(cutoff, ZoneId.systemDefault()));
+    }
+}

app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageMatch.java

@@ -0,0 +1,19 @@
+package stirling.software.saas.payg.lineage;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+import stirling.software.saas.payg.model.ArtifactKind;
+
+/**
+ * Result of a successful lineage lookup: the open job a tool call should join, plus the {@link
+ * ArtifactKind} of the recorded artifact that produced the match (was the matched hash the prior
+ * call's input or its output?) and the job's {@code last_step_at} so the caller can decide whether
+ * to extend the workflow window.
+ *
+ * <p>If a single job has multiple recorded signatures that all match the candidate set (e.g. both
+ * its INPUT and OUTPUT hashes align with what the caller offered), {@code matchedKind} reflects one
+ * of those matches — ordering across same-job-different-kind rows is unspecified. Callers needing
+ * to enumerate every matched kind should query {@link JobLineageStore} directly.
+ */
+public record LineageMatch(UUID jobId, ArtifactKind matchedKind, LocalDateTime jobLastStepAt) {}

app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageSignature.java

@@ -0,0 +1,50 @@
+package stirling.software.saas.payg.lineage;
+
+import java.util.Objects;
+
+/**
+ * A single identity signal for a piece of content. The detector treats {@code (type, value)} as an
+ * opaque equality key — two signatures match if both components are equal. Format:
+ *
+ * <ul>
+ *   <li>{@code type} describes the extraction strategy ({@code "sha256"}, {@code "pdf-id"}, {@code
+ *       "pdf-content-stream-hash"}, etc.).
+ *   <li>{@code value} is the strategy-specific identifier, encoded as a string short enough to fit
+ *       in {@code job_artifact_hash.content_hash} (VARCHAR(128)).
+ * </ul>
+ *
+ * <p>Persisted as the concatenation {@code "{type}:{value}"} so multiple signature types can
+ * coexist on the same {@code job_artifact_hash} table without a separate column.
+ */
+public record LineageSignature(String type, String value) {
+
+    public LineageSignature {
+        Objects.requireNonNull(type, "type");
+        Objects.requireNonNull(value, "value");
+        if (type.isBlank()) {
+            throw new IllegalArgumentException("signature type must not be blank");
+        }
+        if (type.contains(":")) {
+            throw new IllegalArgumentException("signature type must not contain ':'");
+        }
+        if (value.isBlank()) {
+            throw new IllegalArgumentException("signature value must not be blank");
+        }
+    }
+
+    /** Storage form: {@code "type:value"}. */
+    public String asStorageKey() {
+        return type + ":" + value;
+    }
+
+    /** Parses a storage-form key back into a {@code LineageSignature}. */
+    public static LineageSignature fromStorageKey(String key) {
+        Objects.requireNonNull(key, "key");
+        int colon = key.indexOf(':');
+        if (colon <= 0 || colon == key.length() - 1) {
+            throw new IllegalArgumentException(
+                    "Storage key must be of the form 'type:value': " + key);
+        }
+        return new LineageSignature(key.substring(0, colon), key.substring(colon + 1));
+    }
+}

app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageSignatureExtractor.java

@@ -0,0 +1,34 @@
+package stirling.software.saas.payg.lineage;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Set;
+
+/**
+ * Extracts one or more {@link LineageSignature}s from a file. Lineage is decided at the file level:
+ * the extractor sees the whole content and returns whatever identity signals it can pull out.
+ *
+ * <p>The default {@link ByteHashSignatureExtractor} returns a single SHA-256 of the bytes. A future
+ * format-aware extractor (e.g. for PDFs, opening with jpdfium to read {@code /ID[0]} or a
+ * content-stream hash) is just another {@link org.springframework.stereotype.Component} bean
+ * implementing this interface — no changes to the detector or the store. Multiple extractors can be
+ * composed; the detector unions their signatures and matches on any of them.
+ *
+ * <p>Callers materialise upload bodies / response bodies to a {@link Path} (via {@code
+ * TempFileManager}) before extraction so the extractor can stream-hash <em>and</em>, optionally,
+ * parse the file with a format-specific library — both from the same on-disk byte sequence.
+ */
+public interface LineageSignatureExtractor {
+
+    /**
+     * Returns every signature this extractor can derive from {@code file}. Empty set is legal —
+     * meaning "this extractor doesn't know how to fingerprint this content" — and is normal for
+     * format-specific extractors that decline to handle non-matching inputs.
+     */
+    Set<LineageSignature> extract(Path file) throws IOException;
+
+    /**
+     * Short, stable name for logging and diagnostics ({@code "sha256"}, {@code "pdf-id"}, etc.).
+     */
+    String name();
+}

app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java

@@ -1,8 +1,10 @@
 package stirling.software.saas.payg.repository;
 
 import java.time.LocalDateTime;
+import java.util.Collection;
 import java.util.List;
 
+import org.springframework.data.domain.Limit;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
@@ -11,6 +13,7 @@ import org.springframework.stereotype.Repository;
 
 import stirling.software.saas.payg.job.JobArtifactHash;
 import stirling.software.saas.payg.job.JobArtifactHash.JobArtifactHashId;
+import stirling.software.saas.payg.lineage.LineageMatch;
 import stirling.software.saas.payg.model.JobStatus;
 
 @Repository
@@ -18,21 +21,27 @@ public interface JobArtifactHashRepository
         extends JpaRepository<JobArtifactHash, JobArtifactHashId> {
 
     /**
-     * Lineage lookup: find the open job (if any) whose recorded input/output hashes include the
+     * Lineage lookup: find open jobs (owned by {@code userId}, {@code last_step_at > since}) whose
-     * supplied content hash, scoped to one user and the workflow window.
+     * recorded artifacts include any of the supplied storage-form signature keys. Ordered by job
+     * activity recency. Caller passes {@link Limit#of(int)} to bound the result set — for the
+     * single-match hot path use {@code Limit.of(1)} so the DB doesn't materialise unwanted rows.
      */
     @Query(
-            "SELECT j.ownerUserId, h.id.jobId FROM JobArtifactHash h"
+            "SELECT new stirling.software.saas.payg.lineage.LineageMatch("
+                    + " h.id.jobId, h.id.kind, j.lastStepAt)"
+                    + " FROM JobArtifactHash h"
                     + " JOIN ProcessingJob j ON j.id = h.id.jobId"
                     + " WHERE j.ownerUserId = :userId"
                     + " AND j.status = :openStatus"
                     + " AND j.lastStepAt > :since"
-                    + " AND h.id.contentHash = :contentHash")
+                    + " AND h.id.contentHash IN :signatures"
-    List<Object[]> findLineageMatches(
+                    + " ORDER BY j.lastStepAt DESC")
+    List<LineageMatch> findOpenJobsForSignatures(
             @Param("userId") Long userId,
             @Param("openStatus") JobStatus openStatus,
             @Param("since") LocalDateTime since,
-            @Param("contentHash") String contentHash);
+            @Param("signatures") Collection<String> signatures,
+            Limit limit);
 
     /** Prunes rows older than {@code cutoff}; run from a scheduled task. */
     @Modifying

app/saas/src/test/java/stirling/software/saas/payg/lineage/ByteHashSignatureExtractorTest.java

@@ -0,0 +1,80 @@
+package stirling.software.saas.payg.lineage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Set;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class ByteHashSignatureExtractorTest {
+
+    private final ByteHashSignatureExtractor extractor = new ByteHashSignatureExtractor();
+
+    @Test
+    void identicalBytes_produceIdenticalSignatures(@TempDir Path tmp) throws IOException {
+        Path a = tmp.resolve("a.bin");
+        Path b = tmp.resolve("b.bin");
+        byte[] contents = "hello, lineage".getBytes();
+        Files.write(a, contents);
+        Files.write(b, contents);
+
+        Set<LineageSignature> sigsA = extractor.extract(a);
+        Set<LineageSignature> sigsB = extractor.extract(b);
+
+        assertThat(sigsA).isEqualTo(sigsB);
+        assertThat(sigsA).hasSize(1);
+        assertThat(sigsA.iterator().next().type()).isEqualTo("sha256");
+    }
+
+    @Test
+    void differentBytes_produceDifferentSignatures(@TempDir Path tmp) throws IOException {
+        Path a = tmp.resolve("a.bin");
+        Path b = tmp.resolve("b.bin");
+        Files.write(a, "one".getBytes());
+        Files.write(b, "two".getBytes());
+
+        Set<LineageSignature> sigsA = extractor.extract(a);
+        Set<LineageSignature> sigsB = extractor.extract(b);
+
+        assertThat(sigsA).isNotEqualTo(sigsB);
+    }
+
+    @Test
+    void emptyFile_isHashable(@TempDir Path tmp) throws IOException {
+        Path empty = tmp.resolve("empty.bin");
+        Files.write(empty, new byte[0]);
+
+        Set<LineageSignature> sigs = extractor.extract(empty);
+
+        // SHA-256 of the empty string is a well-known constant.
+        assertThat(sigs).hasSize(1);
+        assertThat(sigs.iterator().next().value())
+                .isEqualTo("e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855");
+    }
+
+    @Test
+    void largeFile_streamsWithoutLoadingFullyInMemory(@TempDir Path tmp) throws IOException {
+        // 10 MiB file. The extractor uses a 64 KiB buffer so this should hash without trouble.
+        Path big = tmp.resolve("big.bin");
+        byte[] block = new byte[64 * 1024];
+        try (var out = Files.newOutputStream(big)) {
+            for (int i = 0; i < 10 * 16; i++) {
+                out.write(block);
+            }
+        }
+
+        Set<LineageSignature> sigs = extractor.extract(big);
+
+        assertThat(sigs).hasSize(1);
+        assertThat(sigs.iterator().next().type()).isEqualTo("sha256");
+    }
+
+    @Test
+    void name_isStable() {
+        assertThat(extractor.name()).isEqualTo("sha256");
+    }
+}

app/saas/src/test/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetectorTest.java

@@ -0,0 +1,277 @@
+package stirling.software.saas.payg.lineage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import stirling.software.saas.payg.model.ArtifactKind;
+import stirling.software.saas.payg.model.JobStatus;
+
+/**
+ * Exercises the detector end-to-end against the in-memory store + a fake extractor with
+ * predetermined signatures. Covers the matching rules: same-user, within window, status=OPEN,
+ * multi-signature, and post-record retrieval. The contract is shared with the JPA store; if Redis
+ * (or any future) impl satisfies the same {@link JobLineageStore} interface, these tests are the
+ * canonical "did I get the abstraction right?" check.
+ */
+class DefaultHashLineageDetectorTest {
+
+    private static final Duration WINDOW = Duration.ofMinutes(5);
+
+    private FakeFileSignatureExtractor extractor;
+    private InMemoryJobLineageStore store;
+    private DefaultHashLineageDetector detector;
+
+    @BeforeEach
+    void setUp() {
+        extractor = new FakeFileSignatureExtractor();
+        store = new InMemoryJobLineageStore();
+        detector = new DefaultHashLineageDetector(List.of(extractor), store, WINDOW);
+    }
+
+    @Test
+    void noPriorJobs_returnsEmpty(@TempDir Path tmp) throws IOException {
+        Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc");
+
+        Optional<LineageMatch> match = detector.detect(42L, file);
+
+        assertThat(match).isEmpty();
+    }
+
+    @Test
+    void matchingInputSignature_returnsLineageMatch(@TempDir Path tmp) throws IOException {
+        UUID job = openJobForUser(42L, LocalDateTime.now());
+        recordSignatureForJob(job, "sha256", "abc", ArtifactKind.INPUT);
+        Path file = givenFileWithSignatures(tmp, "next.bin", "sha256", "abc");
+
+        Optional<LineageMatch> match = detector.detect(42L, file);
+
+        assertThat(match).isPresent();
+        assertThat(match.get().jobId()).isEqualTo(job);
+        assertThat(match.get().matchedKind()).isEqualTo(ArtifactKind.INPUT);
+    }
+
+    @Test
+    void matchingOutputSignature_alsoCounts(@TempDir Path tmp) throws IOException {
+        // OCR's output hash matches Compress's input hash — that's the lineage chain.
+        UUID ocr = openJobForUser(42L, LocalDateTime.now());
+        recordSignatureForJob(ocr, "sha256", "ocr-output-hash", ArtifactKind.OUTPUT);
+        Path compressInput =
+                givenFileWithSignatures(tmp, "to-compress.bin", "sha256", "ocr-output-hash");
+
+        Optional<LineageMatch> match = detector.detect(42L, compressInput);
+
+        assertThat(match).isPresent();
+        assertThat(match.get().jobId()).isEqualTo(ocr);
+        assertThat(match.get().matchedKind()).isEqualTo(ArtifactKind.OUTPUT);
+    }
+
+    @Test
+    void differentUser_doesNotMatch(@TempDir Path tmp) throws IOException {
+        UUID job = openJobForUser(/* userA= */ 1L, LocalDateTime.now());
+        recordSignatureForJob(job, "sha256", "abc", ArtifactKind.INPUT);
+        Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc");
+
+        Optional<LineageMatch> match = detector.detect(/* userB= */ 2L, file);
+
+        assertThat(match).isEmpty();
+    }
+
+    @Test
+    void outsideWorkflowWindow_doesNotMatch(@TempDir Path tmp) throws IOException {
+        // Job last touched 10 minutes ago; window is 5 minutes.
+        UUID job = openJobForUser(42L, LocalDateTime.now().minus(Duration.ofMinutes(10)));
+        recordSignatureForJob(job, "sha256", "abc", ArtifactKind.INPUT);
+        Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc");
+
+        Optional<LineageMatch> match = detector.detect(42L, file);
+
+        assertThat(match).isEmpty();
+    }
+
+    @Test
+    void closedJob_doesNotMatch(@TempDir Path tmp) throws IOException {
+        UUID job = UUID.randomUUID();
+        store.registerJob(job, 42L, JobStatus.CLOSED, LocalDateTime.now());
+        recordSignatureForJob(job, "sha256", "abc", ArtifactKind.OUTPUT);
+        Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc");
+
+        Optional<LineageMatch> match = detector.detect(42L, file);
+
+        assertThat(match).isEmpty();
+    }
+
+    @Test
+    void multipleSignatures_matchIfAnyMatches(@TempDir Path tmp) throws IOException {
+        // The job was recorded with one signature (sha256). The incoming file produces TWO
+        // signatures (sha256 + pdf-id). Different sha256 (bytes changed) but same pdf-id → match.
+        UUID job = openJobForUser(42L, LocalDateTime.now());
+        recordSignatureForJob(job, "pdf-id", "stable-uuid", ArtifactKind.OUTPUT);
+
+        Path file =
+                givenFileWithSignatures(
+                        tmp,
+                        "input.pdf",
+                        // Two signatures from the same file:
+                        "sha256",
+                        "differs-because-bytes-changed",
+                        "pdf-id",
+                        "stable-uuid");
+
+        Optional<LineageMatch> match = detector.detect(42L, file);
+
+        assertThat(match).isPresent();
+        assertThat(match.get().jobId()).isEqualTo(job);
+    }
+
+    @Test
+    void mostRecentMatchWins_whenMultipleJobsHaveTheSignature(@TempDir Path tmp)
+            throws IOException {
+        UUID older = openJobForUser(42L, LocalDateTime.now().minus(Duration.ofMinutes(2)));
+        UUID newer = openJobForUser(42L, LocalDateTime.now());
+        recordSignatureForJob(older, "sha256", "shared", ArtifactKind.OUTPUT);
+        recordSignatureForJob(newer, "sha256", "shared", ArtifactKind.OUTPUT);
+        Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "shared");
+
+        Optional<LineageMatch> match = detector.detect(42L, file);
+
+        assertThat(match).isPresent();
+        assertThat(match.get().jobId()).isEqualTo(newer);
+    }
+
+    @Test
+    void record_persistsSignatures(@TempDir Path tmp) throws IOException {
+        UUID job = openJobForUser(42L, LocalDateTime.now());
+        Path file = givenFileWithSignatures(tmp, "output.bin", "sha256", "fresh-hash");
+
+        detector.record(job, file, ArtifactKind.OUTPUT);
+
+        // Now a fresh detect() call against a different file with the same hash should match.
+        Path subsequent = givenFileWithSignatures(tmp, "next.bin", "sha256", "fresh-hash");
+        Optional<LineageMatch> match = detector.detect(42L, subsequent);
+
+        assertThat(match).isPresent();
+        assertThat(match.get().jobId()).isEqualTo(job);
+        assertThat(match.get().matchedKind()).isEqualTo(ArtifactKind.OUTPUT);
+    }
+
+    @Test
+    void extractorThrowing_doesNotBreakRecording(@TempDir Path tmp) throws IOException {
+        // Composite scenario: one of two extractors throws; the other still contributes.
+        FakeFileSignatureExtractor throwing =
+                new FakeFileSignatureExtractor() {
+                    @Override
+                    public Set<LineageSignature> extract(Path file) throws IOException {
+                        throw new IOException(
+                                "synthetic — pretend the file is malformed for this type");
+                    }
+
+                    @Override
+                    public String name() {
+                        return "throwing-extractor";
+                    }
+                };
+        DefaultHashLineageDetector composite =
+                new DefaultHashLineageDetector(List.of(throwing, extractor), store, WINDOW);
+        UUID job = openJobForUser(42L, LocalDateTime.now());
+
+        Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc");
+        composite.record(job, file, ArtifactKind.INPUT);
+
+        // The throwing extractor's signatures didn't land, but the byte-hash one's did, so a
+        // subsequent detect on a matching sha256 hash still finds the job.
+        Path next = givenFileWithSignatures(tmp, "next.bin", "sha256", "abc");
+        Optional<LineageMatch> match = composite.detect(42L, next);
+        assertThat(match).isPresent();
+    }
+
+    @Test
+    void constructor_rejectsEmptyExtractorList() {
+        assertThatThrownBy(() -> new DefaultHashLineageDetector(List.of(), store, WINDOW))
+                .isInstanceOf(IllegalStateException.class);
+    }
+
+    @Test
+    void constructor_rejectsNonPositiveWorkflowWindow() {
+        // Zero window: since == now, `> :since` never matches.
+        assertThatThrownBy(
+                        () ->
+                                new DefaultHashLineageDetector(
+                                        List.of(extractor), store, Duration.ZERO))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("must be positive");
+
+        // Negative window: since == now + |window|, `> :since` only matches the future. Silent
+        // no-match — must fail loud at startup instead.
+        assertThatThrownBy(
+                        () ->
+                                new DefaultHashLineageDetector(
+                                        List.of(extractor), store, Duration.ofMinutes(5).negated()))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("must be positive");
+    }
+
+    // --- helpers --------------------------------------------------------------------------------
+
+    private UUID openJobForUser(long userId, LocalDateTime lastStepAt) {
+        UUID id = UUID.randomUUID();
+        store.registerJob(id, userId, JobStatus.OPEN, lastStepAt);
+        return id;
+    }
+
+    private void recordSignatureForJob(UUID jobId, String type, String value, ArtifactKind kind) {
+        store.record(jobId, Set.of(new LineageSignature(type, value)), kind);
+    }
+
+    private Path givenFileWithSignatures(Path tmp, String filename, String... typesAndValues)
+            throws IOException {
+        // typesAndValues are pairs: type1, value1, type2, value2, ...
+        if (typesAndValues.length % 2 != 0) {
+            throw new IllegalArgumentException("typesAndValues must be in pairs");
+        }
+        Set<LineageSignature> signatures =
+                java.util.stream.IntStream.iterate(0, i -> i < typesAndValues.length, i -> i + 2)
+                        .mapToObj(
+                                i -> new LineageSignature(typesAndValues[i], typesAndValues[i + 1]))
+                        .collect(java.util.stream.Collectors.toSet());
+
+        Path file = tmp.resolve(filename);
+        Files.write(file, ("placeholder bytes for " + filename).getBytes());
+        extractor.setSignaturesFor(file, signatures);
+        return file;
+    }
+
+    /** Test double: returns whatever signatures the test registered for a given file. */
+    private static class FakeFileSignatureExtractor implements LineageSignatureExtractor {
+        private final Map<Path, Set<LineageSignature>> programmed = new HashMap<>();
+
+        void setSignaturesFor(Path file, Set<LineageSignature> sigs) {
+            programmed.put(file, sigs);
+        }
+
+        @Override
+        public Set<LineageSignature> extract(Path file) throws IOException {
+            return programmed.getOrDefault(file, Set.of());
+        }
+
+        @Override
+        public String name() {
+            return "fake";
+        }
+    }
+}

app/saas/src/test/java/stirling/software/saas/payg/lineage/InMemoryJobLineageStore.java

@@ -0,0 +1,90 @@
+package stirling.software.saas.payg.lineage;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+import stirling.software.saas.payg.model.ArtifactKind;
+import stirling.software.saas.payg.model.JobStatus;
+
+/**
+ * In-memory test double for {@link JobLineageStore}. Same interface as the JPA store, plus a {@link
+ * #registerJob} hook so tests can model "this job exists with this user/status/lastStepAt" without
+ * standing up a database.
+ *
+ * <p>The store keeps two maps: one of recorded hash entries, one of registered job state. The
+ * findOpenJobForSignatures impl joins them in-process the same way the JPA query does at the DB.
+ *
+ * <p>Lives in {@code src/test} because it's only useful for unit testing the detector. A future
+ * Redis-backed production impl would live in {@code src/main} alongside {@link JpaJobLineageStore}.
+ */
+public class InMemoryJobLineageStore implements JobLineageStore {
+
+    // All access is synchronized on `this` — a plain HashMap is fine here; ConcurrentHashMap
+    // would duplicate the locking the synchronized methods already provide.
+    private final List<Entry> entries = new ArrayList<>();
+    private final Map<UUID, JobState> jobs = new HashMap<>();
+
+    public synchronized void registerJob(
+            UUID jobId, Long ownerUserId, JobStatus status, LocalDateTime lastStepAt) {
+        jobs.put(jobId, new JobState(ownerUserId, status, lastStepAt));
+    }
+
+    @Override
+    public synchronized void record(
+            UUID jobId, Set<LineageSignature> signatures, ArtifactKind kind) {
+        Instant now = Instant.now();
+        for (LineageSignature sig : signatures) {
+            entries.add(new Entry(jobId, sig.asStorageKey(), kind, now));
+        }
+    }
+
+    @Override
+    public synchronized Optional<LineageMatch> findOpenJobForSignatures(
+            Long userId, Set<LineageSignature> candidates, Duration workflowWindow) {
+        LocalDateTime since = LocalDateTime.now().minus(workflowWindow);
+        Set<String> candidateKeys =
+                candidates.stream()
+                        .map(LineageSignature::asStorageKey)
+                        .collect(java.util.stream.Collectors.toSet());
+
+        return entries.stream()
+                .filter(e -> candidateKeys.contains(e.storageKey()))
+                .map(
+                        e -> {
+                            JobState job = jobs.get(e.jobId());
+                            if (job == null) return null;
+                            if (!userId.equals(job.ownerUserId())) return null;
+                            if (job.status() != JobStatus.OPEN) return null;
+                            if (!job.lastStepAt().isAfter(since)) return null;
+                            return new LineageMatch(e.jobId(), e.kind(), job.lastStepAt());
+                        })
+                .filter(java.util.Objects::nonNull)
+                .max(Comparator.comparing(LineageMatch::jobLastStepAt));
+    }
+
+    @Override
+    public synchronized int pruneOlderThan(Instant cutoff) {
+        int before = entries.size();
+        entries.removeIf(e -> e.recordedAt().isBefore(cutoff));
+        return before - entries.size();
+    }
+
+    /** Test helper — clears all state. */
+    public synchronized void clear() {
+        entries.clear();
+        jobs.clear();
+    }
+
+    private record Entry(UUID jobId, String storageKey, ArtifactKind kind, Instant recordedAt) {}
+
+    private record JobState(Long ownerUserId, JobStatus status, LocalDateTime lastStepAt) {}
+}

app/saas/src/test/java/stirling/software/saas/payg/lineage/LineageSignatureTest.java

@@ -0,0 +1,65 @@
+package stirling.software.saas.payg.lineage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.junit.jupiter.api.Test;
+
+class LineageSignatureTest {
+
+    @Test
+    void storageKey_isTypeAndValueColonJoined() {
+        LineageSignature sig = new LineageSignature("sha256", "abc123");
+
+        assertThat(sig.asStorageKey()).isEqualTo("sha256:abc123");
+    }
+
+    @Test
+    void fromStorageKey_roundTrips() {
+        LineageSignature original = new LineageSignature("pdf-id", "deadbeef-cafe-1234");
+
+        LineageSignature parsed = LineageSignature.fromStorageKey(original.asStorageKey());
+
+        assertThat(parsed).isEqualTo(original);
+    }
+
+    @Test
+    void typeContainingColon_isRejected() {
+        // Otherwise the storage-key parse would be ambiguous.
+        assertThatThrownBy(() -> new LineageSignature("foo:bar", "value"))
+                .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    void valueCanContainColon() {
+        // Values are free-form because they're whatever the extractor produces. The parser only
+        // splits on the FIRST colon, so colons in the value are preserved.
+        LineageSignature sig = new LineageSignature("custom", "a:b:c");
+
+        LineageSignature parsed = LineageSignature.fromStorageKey(sig.asStorageKey());
+
+        assertThat(parsed.value()).isEqualTo("a:b:c");
+    }
+
+    @Test
+    void blankType_isRejected() {
+        assertThatThrownBy(() -> new LineageSignature("", "value"))
+                .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    void blankValue_isRejected() {
+        assertThatThrownBy(() -> new LineageSignature("sha256", ""))
+                .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    void malformedStorageKey_isRejected() {
+        assertThatThrownBy(() -> LineageSignature.fromStorageKey("nocolon"))
+                .isInstanceOf(IllegalArgumentException.class);
+        assertThatThrownBy(() -> LineageSignature.fromStorageKey(":nothingbefore"))
+                .isInstanceOf(IllegalArgumentException.class);
+        assertThatThrownBy(() -> LineageSignature.fromStorageKey("nothingafter:"))
+                .isInstanceOf(IllegalArgumentException.class);
+    }
+}