Add lineage detection primitives: signature extractor, store, detector

Three orthogonal interfaces, each with one production impl. Designed so the
hash algorithm, the storage backend, and the matching policy can each be
swapped independently without changes elsewhere.

LineageSignatureExtractor — given a Path, returns Set<LineageSignature>.
  File-based (not stream-based) so a future PDF-aware extractor can open the
  same file with jpdfium / PDFBox and pull /ID[0] or a content-stream hash
  alongside the byte hash. Multiple extractors compose at the detector
  layer via Spring auto-wiring; the detector unions their results.
  Default impl: ByteHashSignatureExtractor — SHA-256 of file bytes via a
  64 KiB-buffered DigestInputStream. Hardware-accelerated by the JVM.

JobLineageStore — record(jobId, signatures, kind) /
  findOpenJobForSignatures(userId, candidates, window) / pruneOlderThan.
  Knows nothing about storage technology. Production impl JpaJobLineageStore
  backs onto job_artifact_hash via a single joined query. A future
  RedisJobLineageStore is a drop-in.

HashLineageDetector — high-level API. Delegates extraction to the
  configured extractors and storage to the store. Reads
  payg.lineage.workflow-window (default PT5M) from config. Future hook:
  when wallet_policy.auto_group_strategy lookup lands, the detect() method
  short-circuits to empty when the team has opted out.

Schema: content_hash widened from CHAR(64) to VARCHAR(128) so non-SHA-256
signatures fit. Storage form is "type:value" — sha256:hexdigest,
pdf-id:uuid, etc. coexist on the same column.

Tests: InMemoryJobLineageStore + FakeFileSignatureExtractor under
src/test exercise the detector end-to-end without a database. Covers
same-user, within-window, status=OPEN, multi-signature, most-recent-wins,
record+detect round-trip, and extractor-failure isolation.

Stacked PR: branched off payg-i2-document-classifier because the JPA
store + JobArtifactHash entity depend on PR #6460's data model. Once that
merges, this PR rebases to target main.

Author: ConnorYoh

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

@@ -0,0 +1,59 @@
+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.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
+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,102 @@
+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.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}.
+ *
+ * <p>Future: when the per-team {@code wallet_policy.auto_group_strategy} lookup lands (alongside
+ * the wallet policy service), {@link #detect} will short-circuit to {@code Optional.empty()} when
+ * the team has explicitly opted out of auto-grouping. Not in this PR — the wiring slot is the first
+ * line of {@code detect}.
+ */
+@Slf4j
+@Component
+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) {
+        if (Objects.requireNonNull(extractors, "extractors").isEmpty()) {
+            throw new IllegalStateException(
+                    "DefaultHashLineageDetector requires at least one LineageSignatureExtractor"
+                            + " bean — none registered.");
+        }
+        this.extractors = List.copyOf(extractors);
+        this.store = Objects.requireNonNull(store, "store");
+        this.workflowWindow = Objects.requireNonNull(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) throws IOException {
+        Set<LineageSignature> union = new HashSet<>();
+        for (LineageSignatureExtractor extractor : extractors) {
+            try {
+                union.addAll(extractor.extract(file));
+            } catch (IOException e) {
+                // A single extractor failing (e.g. PDF-aware extractor on a malformed PDF) must
+                // not block the byte-hash extractor from contributing. Log and continue.
+                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,78 @@
+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.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import org.springframework.stereotype.Component;
+
+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
+@RequiredArgsConstructor
+public class JpaJobLineageStore implements JobLineageStore {
+
+    private final JobArtifactHashRepository hashRepository;
+
+    @Override
+    public void record(UUID jobId, Set<LineageSignature> signatures, ArtifactKind kind) {
+        Objects.requireNonNull(jobId, "jobId");
+        Objects.requireNonNull(signatures, "signatures");
+        Objects.requireNonNull(kind, "kind");
+        for (LineageSignature signature : signatures) {
+            JobArtifactHash row = new JobArtifactHash();
+            row.setId(new JobArtifactHashId(jobId, signature.asStorageKey(), kind));
+            hashRepository.save(row);
+        }
+    }
+
+    @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)
+                        .collect(Collectors.toList());
+        LocalDateTime since = LocalDateTime.now().minus(workflowWindow);
+
+        List<LineageMatch> matches =
+                hashRepository.findOpenJobsForSignatures(
+                        userId, JobStatus.OPEN, since, storageKeys);
+        return matches.isEmpty() ? Optional.empty() : Optional.of(matches.get(0));
+    }
+
+    @Override
+    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,14 @@
+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.
+ */
+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));
+    }
+}