PAYG: hash-lineage detection primitives (modular extractor / store / detector) (Stirling-Tools#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)

Author: ConnorYoh

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) {}