diff --git a/CHANGELOG.md b/CHANGELOG.md index 1af2f01..ead9960 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,26 @@ may change in any release. See the status note in the README. ## [Unreleased] +The chat's suggestion warm-up stops spending the chat's day on a chat nobody opens, recovers from a failed or +unreachable model within minutes, and the pipeline no longer imports the chat. No migration, and no value to set. + ### Changed +- **The warm-up waits for the chat to be used, and for the changes to pause.** Before the first open since start, a + run's end or a status change prepares nothing; after it, the warm-up starts once `leadgen.chat.suggestion-quiet-window` + (default 30 s) has passed without a further change, so triaging applications one by one costs one call from the + chat's day per pause rather than one per move. The first open after a start, and an open inside the window after a + change, show the catalog's sentences and phrase in the background as any miss does. +- **A failed phrasing is asked once more after a few minutes.** A call that failed, or a chat model that is named but + cannot be built, is left alone for `leadgen.chat.suggestion-retry-after` (default 5 min) instead of until the data + moved, so a local model's cold start no longer leaves the catalog's sentences up until the next run. +- **A miss is phrased in its own language.** Each miss hands over the job it built, so a German miss followed by an + English open is phrased once in each language, and the warm-up no longer reads the data a second time. +- **Follow-ups are no longer stored.** A finished turn's follow-up phrasings stay in memory, so they no longer push the + suggestions out of `chat_suggestion_phrasing`'s 256 rows; rows stored by 0.5.6 age out. +- **The data-change event moved out of the chat.** `chat.suggest.SuggestionDataChanged` is now `event.DataChanged`; + `analytics`, `application` and `archive` import nothing of the chat, and `ChatIsolationTest` checks imports and every + `chat_*` table the migrations create. - **Spring AI 2.0.2, with openai-java held at 4.65.0.** 2.0.2 brings openai-java 4.80.0, and from 4.66.0 on a stopped turn or one past its deadline leaves the model call running (openai/openai-java#1027; 4.96.0 still does). The hold moves from Spring AI to openai-java, so Spring AI's own fixes land: Anthropic streaming chunks carry the thinking @@ -29,6 +47,12 @@ may change in any release. See the status note in the README. 2.98.0 for a commit of its own; Gradle 9.9.0 waits for its `gradle:9.9.0-jdk25` image. openai-java 4.101.0 still fails the four chat-stop tests, so the hold at 4.65.0 stays. +### Fixed + +- **A run finishing during shutdown is no longer logged as not recorded.** The warm-up runs on Boot's + `applicationTaskExecutor` instead of an executor of its own, and a data change after shutdown is dropped instead of + throwing back into the writer. + ## [0.5.7] — 2026-10-31 A maintenance release: dependency updates since 0.5.6, one security fix in the frontend, and a test flake fixed. diff --git a/backend/src/main/java/de/codeministry/leadgen/LeadGenRuntimeHints.java b/backend/src/main/java/de/codeministry/leadgen/LeadGenRuntimeHints.java index 34facc2..32e3efb 100644 --- a/backend/src/main/java/de/codeministry/leadgen/LeadGenRuntimeHints.java +++ b/backend/src/main/java/de/codeministry/leadgen/LeadGenRuntimeHints.java @@ -41,10 +41,8 @@ import de.codeministry.leadgen.chat.suggest.FollowUp; import de.codeministry.leadgen.chat.suggest.Suggestion; import de.codeministry.leadgen.chat.suggest.SuggestionCandidate; -import de.codeministry.leadgen.chat.suggest.SuggestionDataChanged; import de.codeministry.leadgen.chat.suggest.SuggestionScope; import de.codeministry.leadgen.chat.suggest.SuggestionThresholds; -import de.codeministry.leadgen.chat.suggest.SuggestionWarmup; import de.codeministry.leadgen.chat.tools.ApplicationTool; import de.codeministry.leadgen.chat.tools.OfferSearchTool; import de.codeministry.leadgen.chat.tools.PinnedAdvert; @@ -171,10 +169,8 @@ private static void registerChatRecords(RuntimeHints hints) { FollowUp.class, Suggestion.class, SuggestionCandidate.class, - SuggestionDataChanged.class, SuggestionScope.class, SuggestionThresholds.class, - SuggestionWarmup.Job.class, NewConversation.class, ModelPick.class, NewTurn.class, diff --git a/backend/src/main/java/de/codeministry/leadgen/LeadGenerationApplication.java b/backend/src/main/java/de/codeministry/leadgen/LeadGenerationApplication.java index aadb92c..cf346c9 100644 --- a/backend/src/main/java/de/codeministry/leadgen/LeadGenerationApplication.java +++ b/backend/src/main/java/de/codeministry/leadgen/LeadGenerationApplication.java @@ -20,11 +20,14 @@ * Entry point. Scheduling is enabled here because the pipeline runs inside this * process (concept § 10, phase 1) — there is no separate worker container. * - *

Asynchronous execution is enabled for exactly one thing: {@code PackageWorker}, which - * writes and removes package folders after the transaction that asked for it has committed. - * The work is disk and templates rather than a request anybody is waiting on, and Boot's own + *

Asynchronous execution is enabled for {@code PackageWorker}, which writes and removes + * package folders after the transaction that asked for it has committed. The work is disk and + * templates rather than a request anybody is waiting on, and Boot's own * {@code applicationTaskExecutor} runs it — a second pool for one listener would be one more - * thing to size and shut down. + * thing to size and shut down. The chat's {@code SuggestionWarmup} runs on the same executor, + * one model call at a time and for up to the model's timeout; size or bound it with both in mind. + * Its quiet window is timed by the JDK's delay timer, not by Boot's {@code taskScheduler}, whose one + * thread a scheduled ingest pass holds for as long as it runs. */ @SpringBootApplication // Named rather than scanned: a `@WebMvcTest` slice filters scanned records out but keeps diff --git a/backend/src/main/java/de/codeministry/leadgen/analytics/PipelineRunRecorder.java b/backend/src/main/java/de/codeministry/leadgen/analytics/PipelineRunRecorder.java index 5332c00..47bd775 100644 --- a/backend/src/main/java/de/codeministry/leadgen/analytics/PipelineRunRecorder.java +++ b/backend/src/main/java/de/codeministry/leadgen/analytics/PipelineRunRecorder.java @@ -8,8 +8,8 @@ */ package de.codeministry.leadgen.analytics; -import de.codeministry.leadgen.chat.suggest.SuggestionDataChanged; import de.codeministry.leadgen.config.ConfigRegistry; +import de.codeministry.leadgen.event.DataChanged; import de.codeministry.leadgen.filter.FilterStage; import de.codeministry.leadgen.ingest.IngestReport; import de.codeministry.leadgen.score.Judges; @@ -323,7 +323,7 @@ private void write( .update(); } // The counts the empty chat suggests from settle here, not at the start (ISC-529). - events.publishEvent(new SuggestionDataChanged("run " + status)); + events.publishEvent(new DataChanged("run " + status)); } catch (RuntimeException e) { log.error("The run was not recorded: {}", e.getMessage(), e); } @@ -347,7 +347,7 @@ public void complete(int scored, int packaged, boolean digestWritten) { // complete, and a run recorded before this table existed has no row. log.debug("No run was waiting for a batch, so nothing was completed"); } else { - events.publishEvent(new SuggestionDataChanged("run completed")); + events.publishEvent(new DataChanged("run completed")); } } catch (RuntimeException e) { log.error("The batched run was not completed in the history: {}", e.getMessage(), e); diff --git a/backend/src/main/java/de/codeministry/leadgen/application/ApplicationService.java b/backend/src/main/java/de/codeministry/leadgen/application/ApplicationService.java index 4b7fc87..837072d 100644 --- a/backend/src/main/java/de/codeministry/leadgen/application/ApplicationService.java +++ b/backend/src/main/java/de/codeministry/leadgen/application/ApplicationService.java @@ -8,7 +8,7 @@ */ package de.codeministry.leadgen.application; -import de.codeministry.leadgen.chat.suggest.SuggestionDataChanged; +import de.codeministry.leadgen.event.DataChanged; import java.sql.ResultSet; import java.sql.SQLException; import java.time.LocalDate; @@ -174,7 +174,7 @@ private static ApplicationView view(ResultSet rs, int row) throws SQLException { public OpenReport openShortlisted() { int opened = jdbc.sql(OPEN_SHORTLISTED).update(); if (opened > 0) { - events.publishEvent(new SuggestionDataChanged("applications opened")); + events.publishEvent(new DataChanged("applications opened")); } int standing = jdbc.sql(SHORTLIST_STANDING).query(Integer.class).single(); log.info("Board: {} opened at NEW, {} offers on the shortlist", opened, standing); @@ -330,7 +330,7 @@ INSERT INTO application_event (application_id, from_status, to_status, note) .params(applicationId, from == null ? null : from.name(), to.name(), note) .update(); // Every status write passes here, so the empty chat's suggestions follow it (ISC-529). - events.publishEvent(new SuggestionDataChanged("application " + to.name())); + events.publishEvent(new DataChanged("application " + to.name())); } /** diff --git a/backend/src/main/java/de/codeministry/leadgen/archive/ArchiveService.java b/backend/src/main/java/de/codeministry/leadgen/archive/ArchiveService.java index 1087d52..621ab19 100644 --- a/backend/src/main/java/de/codeministry/leadgen/archive/ArchiveService.java +++ b/backend/src/main/java/de/codeministry/leadgen/archive/ArchiveService.java @@ -9,8 +9,8 @@ package de.codeministry.leadgen.archive; import de.codeministry.leadgen.application.ApplicationStatus; -import de.codeministry.leadgen.chat.suggest.SuggestionDataChanged; import de.codeministry.leadgen.config.ConfigRegistry; +import de.codeministry.leadgen.event.DataChanged; import java.time.LocalDate; import java.util.Arrays; import java.util.Collection; @@ -238,7 +238,7 @@ public ArchiveResult setArchived(Collection offerIds, boolean archived) { int reset = jdbc.sql(RESET_TO_NEW).param(ids).update(); if (reset > 0) { log.info("{} restored offers went back to NEW", reset); - events.publishEvent(new SuggestionDataChanged("restored to NEW")); + events.publishEvent(new DataChanged("restored to NEW")); } } return result; diff --git a/backend/src/main/java/de/codeministry/leadgen/chat/InProcess.java b/backend/src/main/java/de/codeministry/leadgen/chat/InProcess.java new file mode 100644 index 0000000..f9ee9c1 --- /dev/null +++ b/backend/src/main/java/de/codeministry/leadgen/chat/InProcess.java @@ -0,0 +1,29 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright 2026 Marcello Muscara (codeministry) + * + * Licensed under the Apache License, Version 2.0. You may obtain a copy of the + * License at http://www.apache.org/licenses/LICENSE-2.0 + */ +package de.codeministry.leadgen.chat; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * A record of the chat that stays inside the process: it is handed from one bean to another and + * never reaches Jackson, so no runtime hint is kept for it (ISC-542). + * + *

Every other record under {@code chat} is read or written as JSON by reflection, and + * {@code LeadGenRuntimeHintsTest} fails for one without a hint. This marker is the one way out of + * that rule, and it says so where the type is declared rather than in a list inside the test: a + * record that later goes into a response or a stored column must lose the marker and gain a hint. + */ +@Documented +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.TYPE) +public @interface InProcess {} diff --git a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/Phrasing.java b/backend/src/main/java/de/codeministry/leadgen/chat/suggest/Phrasing.java index 783e856..ccbac3d 100644 --- a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/Phrasing.java +++ b/backend/src/main/java/de/codeministry/leadgen/chat/suggest/Phrasing.java @@ -13,6 +13,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import de.codeministry.leadgen.chat.ChatBudget; import de.codeministry.leadgen.chat.ChatCapability; +import de.codeministry.leadgen.config.ConfigProperties; import de.codeministry.leadgen.config.ConfigRegistry; import de.codeministry.leadgen.config.model.PipelineConfig; import de.codeministry.leadgen.llm.Answers; @@ -21,6 +22,7 @@ import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.time.Instant; import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; @@ -50,15 +52,17 @@ * *

Looked up apart from asked (ISC-527). {@link #stored} never calls a model, so the empty * chat answers at once; {@link #attempt} is the call itself, which {@link SuggestionWarmup} makes off - * the request thread. {@link #phrase} does both in a row and stays for the follow-ups, which are - * asked once a turn has finished rather than while the chat opens. + * the request thread. {@link #phrase} is the follow-ups' path, asked once a turn has finished rather + * than while the chat opens: a lookup in memory, else a call now. * *

Once per snapshot. The caller's key names the data the sentences were found in; the * model, the language and the sentences themselves are added here, so a changed count, a switched * model or the other language is a new key and a reopened chat with none of those is a map - * lookup. The map is cleared when it grows past {@link #CACHE_LIMIT}: its keys come from data - * snapshots, and an old snapshot is never asked for again. Behind the map sits - * {@link PhrasingStore}, so a restart costs a table read rather than a model call (ISC-531). + * lookup. Each map is cleared when it grows past {@link #CACHE_LIMIT}: its keys come from data + * snapshots, and an old snapshot is never asked for again. The suggestions' phrasings sit in front + * of {@link PhrasingStore}, so for them a restart costs a table read rather than a model call + * (ISC-531). The follow-ups' sit in a map of their own and nowhere else (ISC-539): a finished turn's + * keys are not asked for again, and sharing one map would let them clear the suggestions' out. * *

A phrasing may change the words, not the numbers. A sentence whose digits differ from * the catalog's is dropped for the catalog's, so no count and no id reaches the screen that the @@ -84,23 +88,37 @@ class Phrasing { private static final String TEMPLATE = load(); + /** The suggestions' phrasings, in front of the table. */ private final Map> cache = new ConcurrentHashMap<>(); + /** The follow-ups' phrasings, in memory only (ISC-539). */ + private final Map> followUps = new ConcurrentHashMap<>(); + /** - * Keys whose call failed: not asked again until the data, the model or the language moves the - * key (ISC-532). A refused take or a missing model is not kept here, since neither sends anything. + * Keys whose call failed, or whose named model could not be built, and the instant each may be + * asked again: {@code leadgen.chat.suggestion-retry-after} later (ISC-535). Not retried by + * itself — the next open or warm-up after that instant asks once more — so a model that is down + * costs one call per key and window, never a loop (ISC-532). A refused take is not kept here, + * since it sends nothing and the next day answers it. At {@link #CACHE_LIMIT} the passed windows + * go first and then the oldest, never the whole map: clearing it would lift every window at + * once and send a burst of calls to a model that is still down. */ - private final Set failed = ConcurrentHashMap.newKeySet(); + private final Map skippedUntil = new ConcurrentHashMap<>(); private final ChatCapability capability; private final ChatModels chatModels; private final ConfigRegistry config; private final ChatBudget budget; private final PhrasingStore store; + private final ConfigProperties properties; /** - * The sentences as the reader sees them, in the order given: the stored phrasing, else one - * asked for now, else the catalog's. + * The follow-ups' path: the sentences as the reader sees them, in the order given — the + * phrasing kept in memory, else one asked for now, else the catalog's. + * + *

Kept in memory only, never in the table (ISC-539): a follow-up belongs to one finished turn + * and is never asked for again, so a stored row would only push the suggestions' rows out of + * the table's bound. * * @param snapshot what the sentences were derived from; equal snapshots share one phrasing * @param sentences the catalog's sentences by key, in the reader's language @@ -110,8 +128,8 @@ Map phrase(String snapshot, Map sentences, Strin return sentences; } return key(snapshot, sentences, language) - .flatMap(this::stored) - .or(() -> attempt(snapshot, sentences, language)) + .map(followUps::get) + .or(() -> attempt(snapshot, sentences, language, false)) .orElse(sentences); } @@ -141,32 +159,38 @@ Optional> stored(String key) { e.toString()); return Optional.empty(); } - found.ifPresent(phrased -> remember(key, phrased)); + found.ifPresent(phrased -> remember(cache, key, phrased)); return found; } - /** The language of the newest stored phrasing, if the table holds any. */ - Optional newestLanguage() { - try { - return store.newestLanguage(); - } catch (RuntimeException e) { - log.warn("The stored suggestion phrasings could not be read: {}", e.toString()); - return Optional.empty(); - } - } - /** - * Asks the chat's model now and keeps what it answered. Empty when there is no model, the - * chat's day is spent or the call fails; the caller shows the catalog's sentences then. + * Asks the chat's model now and keeps what it answered, in memory and in the table, or a failure + * under the retry window (ISC-535). Empty when there is no model, the chat's day is spent or the + * call fails; the caller shows the catalog's sentences then. */ Optional> attempt(String snapshot, Map sentences, String language) { + return attempt(snapshot, sentences, language, true); + } + + private Optional> attempt( + String snapshot, Map sentences, String language, boolean persist) { Optional modelName = capability.model(); PipelineConfig.Llm llm = config.snapshot().application().llm(); - Optional model = modelName.flatMap(name -> chatModels.of(llm, name)); - if (model.isEmpty()) { + if (modelName.isEmpty()) { return Optional.empty(); } String key = key(modelName.get(), snapshot, sentences, language); + Optional model = chatModels.of(llm, modelName.get()); + if (model.isEmpty()) { + if (persist) { + log.warn("The chat model is named but could not be built; the catalog's sentences are shown"); + skip(key); + } else { + // Said here too: someone who only reads conversations would otherwise never learn it. + log.warn("The chat model is named but could not be built; the follow-ups show the catalog's sentences"); + } + return Optional.empty(); + } if (!budget.take()) { return Optional.empty(); } @@ -177,10 +201,11 @@ Optional> attempt(String snapshot, Map sente log.warn( "The chat model could not phrase the suggestions; the catalog's sentences are shown: {}", e.toString()); - if (failed.size() >= CACHE_LIMIT) { - failed.clear(); + // A follow-up key is never asked again, so it is not kept: it would only crowd the + // suggestions' entries out of the skip map's bound. + if (persist) { + skip(key); } - failed.add(key); return Optional.empty(); } Map phrased = new LinkedHashMap<>(); @@ -189,7 +214,11 @@ Optional> attempt(String snapshot, Map sente phrased.put(item, acceptable(catalog, candidate) ? candidate.strip() : catalog); }); Map result = Collections.unmodifiableMap(phrased); - remember(key, result); + if (!persist) { + remember(followUps, key, result); + return Optional.of(result); + } + remember(cache, key, result); try { store.save(key, snapshot, language, modelName.get(), sentences, result); } catch (RuntimeException e) { @@ -198,16 +227,41 @@ Optional> attempt(String snapshot, Map sente return Optional.of(result); } - /** Whether the call for {@code key} failed already; such a key is not asked again. */ - boolean failed(String key) { - return failed.contains(key); + /** + * Whether {@code key} failed, or found its model unbuildable, less than the retry window ago; + * such a key is neither asked nor queued until the window has passed. + */ + boolean skipped(String key) { + Instant until = skippedUntil.get(key); + if (until == null) { + return false; + } + if (Instant.now().isBefore(until)) { + return true; + } + skippedUntil.remove(key, until); + return false; + } + + private void skip(String key) { + Instant now = Instant.now(); + if (skippedUntil.size() >= CACHE_LIMIT) { + skippedUntil.values().removeIf(until -> !now.isBefore(until)); + } + if (skippedUntil.size() >= CACHE_LIMIT) { + // One window for every key, so the earliest end is the oldest skip. + skippedUntil.entrySet().stream() + .min(Map.Entry.comparingByValue()) + .ifPresent(oldest -> skippedUntil.remove(oldest.getKey(), oldest.getValue())); + } + skippedUntil.put(key, now.plus(properties.chat().suggestionRetryAfter())); } - private void remember(String key, Map phrased) { - if (cache.size() >= CACHE_LIMIT) { - cache.clear(); + private static void remember(Map> map, String key, Map phrased) { + if (map.size() >= CACHE_LIMIT) { + map.clear(); } - cache.put(key, phrased); + map.put(key, phrased); } private static String key(String model, String snapshot, Map sentences, String language) { diff --git a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/PhrasingStore.java b/backend/src/main/java/de/codeministry/leadgen/chat/suggest/PhrasingStore.java index 6b06a47..2daa0b8 100644 --- a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/PhrasingStore.java +++ b/backend/src/main/java/de/codeministry/leadgen/chat/suggest/PhrasingStore.java @@ -70,13 +70,6 @@ WHERE cache_key IN (SELECT cache_key FROM chat_suggestion_phrasing """).param(KEPT).update(); } - /** The language of the newest row: what the warm-up uses before the first open since start. */ - Optional newestLanguage() { - return jdbc.sql("SELECT language FROM chat_suggestion_phrasing ORDER BY created_at DESC LIMIT 1") - .query(String.class) - .optional(); - } - static String hash(String key) { try { return HexFormat.of() diff --git a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionDataChanged.java b/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionDataChanged.java deleted file mode 100644 index ccc24db..0000000 --- a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionDataChanged.java +++ /dev/null @@ -1,22 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * - * Copyright 2026 Marcello Muscara (codeministry) - * - * Licensed under the Apache License, Version 2.0. You may obtain a copy of the - * License at http://www.apache.org/licenses/LICENSE-2.0 - */ -package de.codeministry.leadgen.chat.suggest; - -/** - * The data the empty chat's suggestions are found in has moved: a run ended or an application - * changed status (ISC-529). Published by whoever writes {@code pipeline_run}'s end or an - * {@code application_event}; {@link SuggestionWarmup} prepares the unpinned suggestions for the - * next open once the write has committed. - * - *

A writer that forgets to publish costs only the warm-up: the next open misses, answers the - * catalog at once and phrases in the background (ISC-528). - * - * @param writer who wrote, for the log - */ -public record SuggestionDataChanged(String writer) {} diff --git a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionService.java b/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionService.java index 03b89d2..b54c735 100644 --- a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionService.java +++ b/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionService.java @@ -16,6 +16,7 @@ import java.util.Optional; import java.util.concurrent.atomic.AtomicReference; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.jdbc.core.simple.JdbcClient; import org.springframework.stereotype.Service; @@ -29,6 +30,7 @@ * meets no data trigger yields only the evergreen pair, each with its number, so the empty chat * always has a floor the data supports. */ +@Slf4j @Service @RequiredArgsConstructor public class SuggestionService { @@ -70,11 +72,13 @@ public List candidates(SuggestionScope scope) { /** * What the empty chat shows: the first {@link #SHOWN} candidates, as the stored phrasing in * {@code language} when there is one and as the catalog's sentences otherwise. Never waits on a - * model (ISC-527): a miss is handed to {@link SuggestionWarmup}, and the next open finds it. + * model (ISC-527): a miss is handed to {@link SuggestionWarmup} as the job this request built, + * and the next open finds it (ISC-538). Every open arms the warm-up for data changes (ISC-536). */ public List suggestions(SuggestionScope scope, String language) { SuggestionScope effective = scope == null ? SuggestionScope.NONE : scope; lastLanguage.set(language); + warmup.opened(); List shown = shown(effective); Map sentences = sentences(shown, language); String snapshot = snapshot(effective); @@ -84,10 +88,17 @@ public List suggestions(SuggestionScope scope, String language) { Optional> stored = phrasing.stored(key.get()); if (stored.isPresent()) { texts = stored.get(); - } else if (effective.equals(SuggestionScope.NONE)) { - warmup.unpinned(); + } else if (phrasing.skipped(key.get())) { + // Failed, or its model could not be built, a moment ago: the catalog until the + // retry window has passed, and nothing queued meanwhile (ISC-535). + log.debug("Suggestion key skipped until its retry window passes; the catalog answers"); } else { - warmup.pinned(new SuggestionWarmup.Job(key.get(), snapshot, sentences, language)); + SuggestionWarmup.Job job = new SuggestionWarmup.Job(key.get(), snapshot, sentences, language); + if (effective.equals(SuggestionScope.NONE)) { + warmup.missed(job); + } else { + warmup.pinned(job); + } } } Map shownTexts = texts; @@ -99,7 +110,8 @@ public List suggestions(SuggestionScope scope, String language) { /** * The unpinned suggestions as they stand now, in the language the last open asked in, ready to - * be phrased; empty when there is nothing to phrase or no chat model to phrase it. + * be phrased after a data change; empty when there is nothing to phrase or no chat model to + * phrase it. A miss never comes here: it hands over the job it already built (ISC-538). */ Optional unpinnedJob() { String language = language(); @@ -113,12 +125,12 @@ Optional unpinnedJob() { } /** - * The language the last open asked in; before the first open since start, the newest stored - * phrasing's, and English when nothing is stored. + * The language the last open asked in. A data change prepares nothing before the first open + * since start (ISC-536), and that open sets it, so English is only a guard. */ String language() { String last = lastLanguage.get(); - return last != null ? last : phrasing.newestLanguage().orElse(Catalog.ENGLISH); + return last != null ? last : Catalog.ENGLISH; } private List shown(SuggestionScope scope) { diff --git a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionWarmup.java b/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionWarmup.java index c653a3c..98f069d 100644 --- a/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionWarmup.java +++ b/backend/src/main/java/de/codeministry/leadgen/chat/suggest/SuggestionWarmup.java @@ -8,15 +8,23 @@ */ package de.codeministry.leadgen.chat.suggest; +import de.codeministry.leadgen.chat.InProcess; +import de.codeministry.leadgen.config.ConfigProperties; +import de.codeministry.leadgen.event.DataChanged; +import java.time.Duration; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.Map; import java.util.Optional; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.task.TaskExecutor; +import org.springframework.core.task.TaskRejectedException; import org.springframework.stereotype.Component; import org.springframework.transaction.event.TransactionPhase; import org.springframework.transaction.event.TransactionalEventListener; @@ -28,117 +36,248 @@ * the catalog's sentences, and hands the miss here; the next open finds the phrasing (ISC-527, * ISC-528). * - *

Two kinds of work. The unpinned suggestions are one slot: asking for them while a call - * runs marks them due once more, and that one more is computed from the data as it is then, so any - * number of changes during a slow call cost a single further call for the newest snapshot. A pinned - * context cannot be prepared ahead and carries its own sentences; those wait in a de-duplicated set - * of at most {@link #PINNED_LIMIT}, and one more is dropped rather than queued, since its next open - * asks again. + *

Three kinds of work. A miss hands over the job its request built: snapshot, sentences + * and the language it was asked in, and the worker phrases exactly that without reading the data + * again (ISC-538). An unpinned miss is kept as the newest for its language, so a German miss and an + * English open after it are each phrased in their own language; a pinned one waits in a set + * de-duplicated by key, of at most {@link #PINNED_LIMIT}, and one more is dropped rather than + * queued, since its next open asks again. A data change instead marks one slot due, "compute the + * unpinned suggestions again", in the last open's language; any number of changes during a slow + * call cost a single further call for the newest snapshot. * - *

One worker thread. A local model answers one request at a time anyway, and a second - * call in flight would only race the first for the chat's day. + *

A data change costs nothing until the chat is used, and once per quiet spell after + * that. Before the first open since start, a run's end or a status change is dropped: nobody + * may ever look, and each warm-up is a call from the chat's day (ISC-536). After it, a change + * starts the warm-up only once {@code leadgen.chat.suggestion-quiet-window} has passed without a + * further one, so triaging a hundred applications one by one costs one call, not a hundred + * (ISC-537). + * + *

One phrasing at a time, on Boot's executor. A local model answers one request at a time + * anyway, and a second call in flight would only race the first for the chat's day. The + * {@code running} flag holds that, not a pool of one: the work runs on Boot's + * {@code applicationTaskExecutor}, as {@code LeadGenerationApplication} asks of every background + * task; nothing here owns a thread (ISC-540). + * + *

The quiet window waits on the JDK's delay timer, not on Boot's {@code taskScheduler}. + * That scheduler has one thread, and a scheduled ingest pass holds it for as long as the pass runs: + * a window waiting behind it passed only when the pass ended, and every unpinned miss in between + * stayed unphrased. {@link CompletableFuture#delayedExecutor} times the window on a timer the JDK + * shares and runs the short hand-over to {@link #start()} on its common pool; the phrasing itself + * still runs on Boot's executor. + * + *

Quiet after shutdown. A run can finish while the context closes, and its data change + * reaches the listener synchronously; once {@link #destroy()} ran, such a trigger is dropped, and a + * refusal from the executor never travels back to the writer (ISC-540). */ @Slf4j @Component public class SuggestionWarmup implements DisposableBean { - /** Enough for the pins of a working session, small against the chat's 200 calls a day. */ + /** Enough for the misses and pins of a working session, small against the chat's 200 calls a day. */ static final int PINNED_LIMIT = 16; /** One phrasing to make: the key it is kept under and what the model is asked to phrase. */ + @InProcess public record Job(String key, String snapshot, Map sentences, String language) {} private final Phrasing phrasing; private final ObjectProvider suggestions; - private final ExecutorService worker = Executors.newSingleThreadExecutor( - Thread.ofVirtual().name("suggestion-warmup").factory()); + private final TaskExecutor executor; + private final Executor afterQuietWindow; private final Object lock = new Object(); - private final Map pinned = new LinkedHashMap<>(); + + /** Pinned misses by key, in arrival order. */ + private final Map pinnedJobs = new LinkedHashMap<>(); + + /** Unpinned misses by language, the newest per language: an older one is a snapshot already gone. */ + private final Map unpinnedMisses = new LinkedHashMap<>(); + + /** Whether a data change asks for the unpinned suggestions to be computed again and phrased. */ private boolean unpinnedDue; + + private boolean armed; + private CompletableFuture pendingStart; + + /** Which scheduled start is the current one; an older one that already fired does nothing. */ + private long generation; + private boolean running; + private boolean closed; - SuggestionWarmup(Phrasing phrasing, ObjectProvider suggestions) { + SuggestionWarmup( + Phrasing phrasing, + ObjectProvider suggestions, + @Qualifier("applicationTaskExecutor") TaskExecutor executor, + ConfigProperties properties) { this.phrasing = phrasing; this.suggestions = suggestions; + this.executor = executor; + Duration quietWindow = properties.chat().suggestionQuietWindow(); + this.afterQuietWindow = CompletableFuture.delayedExecutor(quietWindow.toNanos(), TimeUnit.NANOSECONDS); } /** * A run ended or an application changed status: the unpinned suggestions are prepared for the - * next open (ISC-529). After the commit, so a rolled-back write prepares nothing, and outside a - * transaction at once ({@code PipelineRunRecorder.complete} runs in none). Only marks the work - * due: the writer's thread never waits on the model. + * next open (ISC-529) — if the chat was opened since start (ISC-536), and once the changes have + * paused for the quiet window (ISC-537). After the commit, so a rolled-back write prepares + * nothing, and outside a transaction at once ({@code PipelineRunRecorder.complete} runs in + * none). Only schedules: the writer's thread never waits on the model. */ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true) - void dataChanged(SuggestionDataChanged event) { - log.debug("Suggestion data changed by {}; the unpinned suggestions are prepared", event.writer()); - unpinned(); + void dataChanged(DataChanged event) { + try { + synchronized (lock) { + if (!armed || closed) { + log.debug("Suggestion data changed by {}; the chat was not opened since start", event.writer()); + return; + } + log.debug("Suggestion data changed by {}; the unpinned suggestions are prepared", event.writer()); + scheduleQuietWindow(); + } + } catch (RuntimeException e) { + // The writer's thread: whatever went wrong here is the warm-up's, never the write's. + log.warn("The suggestion warm-up could not be scheduled after a data change: {}", e.toString()); + } } - /** Marks the unpinned suggestions due; they are computed when the worker gets to them. */ - void unpinned() { + /** The chat was opened: from now on a data change prepares the next open (ISC-536). */ + void opened() { synchronized (lock) { - unpinnedDue = true; - start(); + armed = true; } } - /** Queues one pinned context's phrasing, unless it waits already or the set is full. */ - void pinned(Job job) { + /** + * An unpinned open missed: keep the job its request built — its snapshot, its sentences, its + * language — as the newest for that language, so a German and an English miss are each phrased + * in their own language (ISC-538). Phrased at once, unless changes are still arriving: then it + * waits for the quiet window like they do, and a reader who reopens the chat after every move + * costs one call per pause, not one per move (ISC-537). + */ + void missed(Job job) { synchronized (lock) { - if (pinned.containsKey(job.key())) { - return; + unpinnedMisses.put(job.language(), job); + if (pendingStart == null) { + start(); } - if (pinned.size() >= PINNED_LIMIT) { - log.debug( - "{} pinned suggestion phrasings wait already; this one is asked again on its next open", - PINNED_LIMIT); - return; + } + } + + /** + * A pinned open missed: queue the job its request built, unless that key waits already or the set + * is full, and make sure the worker runs (ISC-528). + */ + void pinned(Job job) { + synchronized (lock) { + if (!pinnedJobs.containsKey(job.key())) { + if (pinnedJobs.size() >= PINNED_LIMIT) { + log.debug( + "{} pinned suggestion phrasings wait already; this one is asked again on its next open", + PINNED_LIMIT); + } else { + pinnedJobs.put(job.key(), job); + } } - pinned.put(job.key(), job); + // Also when the key waits already: a start the executor refused left it there. start(); } } - /** Whether nothing is running or waiting: what a caller that waits for the warm-up reads. */ + /** + * Whether nothing is running, waiting or scheduled: what a caller that waits for the warm-up + * reads. A start the quiet window still holds back counts as work. + */ public boolean idle() { synchronized (lock) { - return !running; + return !running && pendingStart == null; } } @Override public void destroy() { - worker.shutdownNow(); + synchronized (lock) { + closed = true; + unpinnedDue = false; + pinnedJobs.clear(); + unpinnedMisses.clear(); + if (pendingStart != null) { + pendingStart.cancel(false); + pendingStart = null; + } + } + } + + /** + * Under {@link #lock}: (re)starts the quiet window, superseding a start still pending. The old + * start is let go before the new one is asked for, so a refusal never leaves a cancelled one + * behind as pending. + */ + private void scheduleQuietWindow() { + long current = ++generation; + if (pendingStart != null) { + pendingStart.cancel(false); + pendingStart = null; + } + pendingStart = CompletableFuture.runAsync(() -> quietWindowPassed(current), afterQuietWindow); + } + + /** + * The changes paused for the quiet window: the unpinned suggestions are computed again, and the + * misses that waited for it are phrased. A start a newer change superseded does nothing, even when + * it fired before its cancel reached it. Only a data change schedules a window, so a window that + * passes always asks for the recompute. + */ + private void quietWindowPassed(long scheduled) { + synchronized (lock) { + if (scheduled != generation) { + return; + } + pendingStart = null; + unpinnedDue = true; + start(); + } } /** Under {@link #lock}. */ private void start() { - if (!running) { - running = true; - worker.execute(this::drain); + if (running || closed) { + return; + } + running = true; + try { + executor.execute(this::drain); + } catch (TaskRejectedException e) { + running = false; + log.warn( + "The suggestion warm-up was refused by the task executor; the next trigger asks again: {}", + e.toString()); } } private void drain() { while (true) { - boolean unpinnedNow; Job next = null; + boolean recompute = false; synchronized (lock) { - unpinnedNow = unpinnedDue; - if (unpinnedNow) { + if (closed) { + running = false; + return; + } + if (unpinnedDue) { unpinnedDue = false; - } else if (!pinned.isEmpty()) { - Iterator first = pinned.values().iterator(); - next = first.next(); - first.remove(); + recompute = true; + } else if (!unpinnedMisses.isEmpty() && pendingStart == null) { + next = take(unpinnedMisses); + } else if (!pinnedJobs.isEmpty()) { + next = take(pinnedJobs); } else { running = false; return; } } try { - Optional job = unpinnedNow ? suggestions.getObject().unpinnedJob() : Optional.of(next); + Optional job = recompute ? suggestions.getObject().unpinnedJob() : Optional.of(next); job.ifPresent(this::phrase); } catch (RuntimeException e) { log.warn("A suggestion phrasing failed; the catalog's sentences stay: {}", e.toString()); @@ -146,8 +285,15 @@ private void drain() { } } + private static Job take(Map queue) { + Iterator first = queue.values().iterator(); + Job job = first.next(); + first.remove(); + return job; + } + private void phrase(Job job) { - if (phrasing.failed(job.key()) || phrasing.stored(job.key()).isPresent()) { + if (phrasing.skipped(job.key()) || phrasing.stored(job.key()).isPresent()) { return; } phrasing.attempt(job.snapshot(), job.sentences(), job.language()); diff --git a/backend/src/main/java/de/codeministry/leadgen/config/ConfigProperties.java b/backend/src/main/java/de/codeministry/leadgen/config/ConfigProperties.java index c64946e..a277d82 100644 --- a/backend/src/main/java/de/codeministry/leadgen/config/ConfigProperties.java +++ b/backend/src/main/java/de/codeministry/leadgen/config/ConfigProperties.java @@ -63,6 +63,8 @@ public record ConfigProperties( private static final String DEFAULT_INGEST_CRON = "-"; private static final String DEFAULT_CHAT_TURN_TIMEOUT = "PT15M"; private static final String DEFAULT_CHAT_HEARTBEAT = "PT15S"; + private static final String DEFAULT_CHAT_SUGGESTION_RETRY_AFTER = "PT5M"; + private static final String DEFAULT_CHAT_SUGGESTION_QUIET_WINDOW = "PT30S"; /** The one Spring binds. Named, because the record has a second constructor below. */ @ConstructorBinding @@ -80,7 +82,11 @@ public ConfigProperties(String configDir) { Duration.parse(DEFAULT_SCORE_BATCH_POLL_INTERVAL), DEFAULT_INGEST_CRON, new Security(false, List.of()), - new Chat(Duration.parse(DEFAULT_CHAT_TURN_TIMEOUT), Duration.parse(DEFAULT_CHAT_HEARTBEAT))); + new Chat( + Duration.parse(DEFAULT_CHAT_TURN_TIMEOUT), + Duration.parse(DEFAULT_CHAT_HEARTBEAT), + Duration.parse(DEFAULT_CHAT_SUGGESTION_RETRY_AFTER), + Duration.parse(DEFAULT_CHAT_SUGGESTION_QUIET_WINDOW))); } // `packages-dir` and `inbox-dir` used to sit here as well, read by nothing: the packages @@ -115,19 +121,33 @@ public record Security( @DefaultValue List allowedHosts) {} /** - * {@code leadgen.chat.*}: the two process-level clocks of a chat turn. Neither is in + * {@code leadgen.chat.*}: the process-level clocks of the chat. None is in * {@code pipeline.yaml}, because a test has to shorten them for one context and a reload must - * not change the deadline of a stream that is already open. + * not change the deadline of a stream that is already open; no deployment is expected to set + * them, so none is in {@code .env.example} either. * * @param turnTimeout how long one turn may take from its first model call to its last event; * {@code ChatTurnService} stops the turn there and {@code ChatController} gives the stream a * little longer, so the reader still gets the turn's {@code error} * @param heartbeat how often a running turn's stream carries an SSE comment, well below any * proxy's idle timeout, so a model silent over a long prompt is not cut off + * @param suggestionRetryAfter how long a suggestion phrasing that failed, or a chat model that + * could not be built, is left alone before the next open or warm-up asks once more + * (ISC-535); long enough that a model that is down costs one call per key and window, short + * enough that a cold start that timed out is over by the next try + * @param suggestionQuietWindow how long data changes must pause before the unpinned + * suggestions are prepared for the next open (ISC-537), so a burst of status moves costs one + * call from the chat's day rather than one per move */ public record Chat( @NotNull @DefaultValue(DEFAULT_CHAT_TURN_TIMEOUT) Duration turnTimeout, - @NotNull @DefaultValue(DEFAULT_CHAT_HEARTBEAT) Duration heartbeat) {} + @NotNull @DefaultValue(DEFAULT_CHAT_HEARTBEAT) Duration heartbeat, + + @NotNull @DefaultValue(DEFAULT_CHAT_SUGGESTION_RETRY_AFTER) + Duration suggestionRetryAfter, + + @NotNull @DefaultValue(DEFAULT_CHAT_SUGGESTION_QUIET_WINDOW) + Duration suggestionQuietWindow) {} } diff --git a/backend/src/main/java/de/codeministry/leadgen/event/DataChanged.java b/backend/src/main/java/de/codeministry/leadgen/event/DataChanged.java new file mode 100644 index 0000000..a755967 --- /dev/null +++ b/backend/src/main/java/de/codeministry/leadgen/event/DataChanged.java @@ -0,0 +1,22 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright 2026 Marcello Muscara (codeministry) + * + * Licensed under the Apache License, Version 2.0. You may obtain a copy of the + * License at http://www.apache.org/licenses/LICENSE-2.0 + */ +package de.codeministry.leadgen.event; + +/** + * The data the working set is read from has moved: a run ended or an application changed status. + * Published by whoever writes {@code pipeline_run}'s end or an {@code application_event}; a + * listener that needs the commit listens after it. + * + *

Today the one listener is the chat's suggestion warm-up (ISC-529). A writer that forgets to + * publish costs only that warm-up: the next open misses, answers the catalog at once and phrases in + * the background (ISC-528). + * + * @param writer who wrote, for the log + */ +public record DataChanged(String writer) {} diff --git a/backend/src/main/java/de/codeministry/leadgen/event/package-info.java b/backend/src/main/java/de/codeministry/leadgen/event/package-info.java new file mode 100644 index 0000000..b29cad8 --- /dev/null +++ b/backend/src/main/java/de/codeministry/leadgen/event/package-info.java @@ -0,0 +1,19 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright 2026 Marcello Muscara (codeministry) + * + * Licensed under the Apache License, Version 2.0. You may obtain a copy of the + * License at http://www.apache.org/licenses/LICENSE-2.0 + */ + +/** + * Facts a writer publishes and no consumer owns. + * + *

A record here says that something happened — a run ended, an application changed status — + * and nothing about who listens. The writer imports this package and nothing of its listeners, so + * a listener can be removed without touching a writer: the chat, which ISC-422 keeps removable, + * prepares its suggestions from {@link de.codeministry.leadgen.event.DataChanged} without the + * pipeline knowing the chat exists (ISC-541). + */ +package de.codeministry.leadgen.event; diff --git a/backend/src/test/java/de/codeministry/leadgen/LeadGenRuntimeHintsTest.java b/backend/src/test/java/de/codeministry/leadgen/LeadGenRuntimeHintsTest.java index 2192105..ce9365d 100644 --- a/backend/src/test/java/de/codeministry/leadgen/LeadGenRuntimeHintsTest.java +++ b/backend/src/test/java/de/codeministry/leadgen/LeadGenRuntimeHintsTest.java @@ -10,6 +10,7 @@ import static org.assertj.core.api.Assertions.assertThat; +import de.codeministry.leadgen.chat.InProcess; import de.codeministry.leadgen.config.ConfigFixtures; import java.io.IOException; import java.lang.reflect.Method; @@ -199,14 +200,18 @@ void everyMcpRecordIsBoundForJson() { * Finding 7: every record of the chat package is read or written by Jackson through reflection — * the SSE events, the conversation views, the ledger's citations in {@code chat_turn.citations}, * the tool results — and none of it is a call AOT can follow. Scanned rather than listed, so a - * new record without a hint fails here; private records never leave their class and are exempt. + * new record without a hint fails here; private records never leave their class and are exempt, + * and so is a record marked {@link InProcess}, which is handed between beans and never serialised + * (ISC-542). */ @Test void everyChatRecordIsBoundForJson() { RuntimeHints hints = new RuntimeHints(); new LeadGenRuntimeHints().registerHints(hints, getClass().getClassLoader()); - List> records = records("de.codeministry.leadgen.chat"); + List> records = records("de.codeministry.leadgen.chat").stream() + .filter(record -> !record.isAnnotationPresent(InProcess.class)) + .toList(); // Guards the guard: the chat package holds more than a dozen records today. assertThat(records).hasSizeGreaterThanOrEqualTo(12); for (Class record : records) { diff --git a/backend/src/test/java/de/codeministry/leadgen/chat/ChatIsolationTest.java b/backend/src/test/java/de/codeministry/leadgen/chat/ChatIsolationTest.java index a7f7266..034b26b 100644 --- a/backend/src/test/java/de/codeministry/leadgen/chat/ChatIsolationTest.java +++ b/backend/src/test/java/de/codeministry/leadgen/chat/ChatIsolationTest.java @@ -15,7 +15,11 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.List; +import java.util.Set; +import java.util.TreeSet; +import java.util.regex.Matcher; import java.util.regex.Pattern; +import java.util.stream.Collectors; import java.util.stream.Stream; import org.junit.jupiter.api.Test; @@ -26,30 +30,97 @@ *

A source scan rather than a runtime check, because the failure it guards against is a line of * code that has not been written yet: a stage that one day joins {@code chat_turn} or injects the * conversation repository. Only {@code chat} itself and the controller that serves it may name them. + * + *

Two scans (ISC-541). The tables are read from the migrations rather than listed here, so the + * next {@code chat_*} table is guarded the day it is created; and no pipeline code imports the + * chat's package at all, so a writer that wants to tell the chat something publishes a neutral + * fact from {@code event} instead. */ class ChatIsolationTest { private static final Path MAIN = Path.of("src/main/java/de/codeministry/leadgen"); - private static final Pattern CHAT_STATE = Pattern.compile( - "chat_conversation|chat_turn|chat_tool_call|chat_call_budget|chat_spend|ConversationRepository"); + private static final Path MIGRATIONS = Path.of("src/main/resources/db/migration"); + + private static final Pattern CREATED_TABLE = + Pattern.compile("(?i)create\\s+table\\s+(?:if\\s+not\\s+exists\\s+)?(chat_\\w+)"); + + /** Any reference to the chat's package — an import, a static import, a qualified name — outside a comment. */ + private static final Pattern CHAT_REFERENCE = Pattern.compile("\\bde\\.codeministry\\.leadgen\\.chat\\."); + + private static final Pattern COMMENT = Pattern.compile("(?s)/\\*.*?\\*/|//[^\\n]*"); + + /** Beside {@code chat} itself, the code that serves the chat or wires it, and runs no stage. */ + private static final Set MAY_NAME_THE_CHAT = Set.of( + MAIN.resolve("web/ChatController.java"), + // The native-image registry names every reflected type by design. + MAIN.resolve("LeadGenRuntimeHints.java")); + + /** May import the chat's types, though never read its tables: they expose the chat, not a stage. */ + private static final List MAY_IMPORT_THE_CHAT = List.of( + MAIN.resolve("mcp"), + // Validates the chat's prices block of pipeline.yaml at load. + MAIN.resolve("config/ConfigLoader.java")); @Test void noPipelineCodeReadsWhatTheChatWrites() throws IOException { - List offenders; + Pattern chatState = chatState(); + List offenders = pipelineSources() + .filter(file -> chatState.matcher(read(file)).find()) + .map(Path::toString) + .toList(); + assertThat(offenders) + .as("pipeline code naming the chat's tables (" + chatState.pattern() + ") or repository") + .isEmpty(); + } + + @Test + void noPipelineCodeImportsTheChat() throws IOException { + List offenders = pipelineSources() + .filter(file -> MAY_IMPORT_THE_CHAT.stream().noneMatch(file::startsWith)) + .filter(file -> CHAT_REFERENCE + .matcher(COMMENT.matcher(read(file)).replaceAll("")) + .find()) + .map(Path::toString) + .toList(); + assertThat(offenders) + .as("pipeline code referring to de.codeministry.leadgen.chat") + .isEmpty(); + } + + @Test + void everyChatTableIsGuarded() throws IOException { + // Guards the guard: eight chat tables exist today, and an empty scan would pass anything. + assertThat(chatTables()).hasSizeGreaterThanOrEqualTo(8).contains("chat_suggestion_phrasing", "chat_context"); + } + + /** Every table a migration creates under the chat's prefix, and the repository the chat writes through. */ + private static Pattern chatState() throws IOException { + return Pattern.compile(String.join("|", chatTables()) + "|ConversationRepository"); + } + + private static Set chatTables() throws IOException { + Set tables = new TreeSet<>(); + try (Stream files = Files.list(MIGRATIONS)) { + for (Path file : files.filter(f -> f.toString().endsWith(".sql")).toList()) { + Matcher matcher = CREATED_TABLE.matcher(read(file)); + while (matcher.find()) { + tables.add(matcher.group(1).toLowerCase(java.util.Locale.ROOT)); + } + } + } + return tables; + } + + private static Stream pipelineSources() throws IOException { try (Stream files = Files.walk(MAIN)) { - offenders = files.filter(file -> file.toString().endsWith(".java")) + return files + .filter(file -> file.toString().endsWith(".java")) .filter(file -> !file.startsWith(MAIN.resolve("chat"))) - .filter(file -> !file.equals(MAIN.resolve("web/ChatController.java"))) - // The native-image registry names every reflected type by design; it runs no stage. - .filter(file -> !file.equals(MAIN.resolve("LeadGenRuntimeHints.java"))) - .filter(file -> CHAT_STATE.matcher(read(file)).find()) - .map(Path::toString) - .toList(); + .filter(file -> !MAY_NAME_THE_CHAT.contains(file)) + .collect(Collectors.toList()) + .stream(); } - assertThat(offenders) - .as("pipeline code naming the chat's tables or repository") - .isEmpty(); } private static String read(Path file) { diff --git a/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionPhrasingTest.java b/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionPhrasingTest.java index ba67eb9..fff1e5b 100644 --- a/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionPhrasingTest.java +++ b/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionPhrasingTest.java @@ -283,6 +283,12 @@ void followUpsArePhrasedOnceUnderTheChatBudget() { assertThat(second).isEqualTo(first); assertThat(chatCalls()).isEqualTo(1); assertThat(llmCalls()).isZero(); + // Kept in memory only: a follow-up is never asked for again, and a row would crowd the + // suggestions out of the table's bound (ISC-539). + assertThat(jdbc.queryForObject( + "SELECT count(*) FROM chat_suggestion_phrasing WHERE snapshot NOT LIKE 'suggestions %'", + Integer.class)) + .isZero(); } /** Waits for the background phrasing to finish, so no call leaks into the next test. */ diff --git a/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionPrewarmTest.java b/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionPrewarmTest.java index f0cf755..78441a8 100644 --- a/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionPrewarmTest.java +++ b/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionPrewarmTest.java @@ -9,6 +9,7 @@ package de.codeministry.leadgen.chat; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assumptions.assumeThat; import de.codeministry.leadgen.Databases; import de.codeministry.leadgen.analytics.PipelineRunRecorder; @@ -16,11 +17,11 @@ import de.codeministry.leadgen.application.ApplicationStatus; import de.codeministry.leadgen.application.ApplicationUpdate; import de.codeministry.leadgen.chat.suggest.Suggestion; -import de.codeministry.leadgen.chat.suggest.SuggestionDataChanged; import de.codeministry.leadgen.chat.suggest.SuggestionScope; import de.codeministry.leadgen.chat.suggest.SuggestionService; import de.codeministry.leadgen.chat.suggest.SuggestionWarmup; import de.codeministry.leadgen.config.ConfigRegistry; +import de.codeministry.leadgen.event.DataChanged; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; @@ -30,14 +31,20 @@ import java.time.Instant; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; import org.springframework.boot.testcontainers.service.connection.ServiceConnection; import org.springframework.context.ApplicationEventPublisher; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.TaskScheduler; import org.springframework.test.context.DynamicPropertyRegistry; import org.springframework.test.context.DynamicPropertySource; import org.testcontainers.junit.jupiter.Container; @@ -54,6 +61,7 @@ */ @SpringBootTest @Testcontainers +@ExtendWith(OutputCaptureExtension.class) class SuggestionPrewarmTest { private static final ModelStub MODEL = ModelStub.start(); @@ -66,6 +74,16 @@ class SuggestionPrewarmTest { /** Two chunks of five seconds each: a model that takes ten seconds to answer. */ private static final Duration SLOW = Duration.ofSeconds(5); + /** The retry window, shortened from its default of minutes so a case can wait it out (ISC-535). */ + private static final Duration RETRY_AFTER = Duration.ofSeconds(2); + + /** + * The quiet window, shortened from its default of 30 s so a burst can be timed (ISC-537). A + * second rather than less: a status move is a transaction, and on a loaded runner one took longer + * than a 300 ms window, which split a burst the test meant as one. + */ + private static final Duration QUIET = Duration.ofSeconds(1); + /** What "at once" means here: an open, a snapshot query and a map lookup, far below this. */ private static final Duration AT_ONCE = Duration.ofSeconds(1); @@ -80,6 +98,8 @@ class SuggestionPrewarmTest { @DynamicPropertySource static void configuration(DynamicPropertyRegistry registry) { registry.add("leadgen.config-dir", () -> STUBBED.getParent().toString()); + registry.add("leadgen.chat.suggestion-retry-after", RETRY_AFTER::toString); + registry.add("leadgen.chat.suggestion-quiet-window", QUIET::toString); } @Autowired @@ -103,6 +123,10 @@ static void configuration(DynamicPropertyRegistry registry) { @Autowired private JdbcTemplate jdbc; + /** Boot's scheduler, one thread, which the ingest pass, the score poll and the config watch run on. */ + @Autowired + private TaskScheduler scheduler; + @AfterAll static void stop() { MODEL.close(); @@ -261,6 +285,8 @@ void aStatusChangeWarmsTheNextOpenWithoutDelayingTheWrite() { @Test void aFailingWarmUpLeavesTheRunAndTheStatusWriteAsTheyWere() { + // An open first: before one since start, a data change prepares nothing (ISC-536). + suggestions.suggestions(SuggestionScope.NONE, "en"); long application = applications.open(offer(), ApplicationStatus.NEW); awaitIdle(); // A run that started after the last open and waits for its batch: a new snapshot. @@ -281,6 +307,8 @@ void aFailingWarmUpLeavesTheRunAndTheStatusWriteAsTheyWere() { @Test void changesDuringASlowWarmUpCollapseIntoOneMoreForTheNewestSnapshot() { + // An open first: before one since start, a data change prepares nothing (ISC-536). + suggestions.suggestions(SuggestionScope.NONE, "en"); long application = applications.open(offer(), ApplicationStatus.NEW); awaitIdle(); MODEL.reset(); @@ -300,7 +328,7 @@ void changesDuringASlowWarmUpCollapseIntoOneMoreForTheNewestSnapshot() { } awaitIdle(); int afterSix = MODEL.bodies().size(); - events.publishEvent(new SuggestionDataChanged("test, nothing moved")); + events.publishEvent(new DataChanged("test, nothing moved")); awaitIdle(); int afterStored = MODEL.bodies().size(); suggestions.suggestions(SuggestionScope.NONE, lastPhrasedLanguage()); @@ -378,6 +406,172 @@ void aPhrasingThatChangesANumberIsStoredWithTheCatalogSentenceInItsPlace() { assertThat(llmCalls()).isZero(); } + @Test + void aFailedKeyIsAskedOnceMoreAfterItsRetryWindowAndNotBefore() throws InterruptedException { + // Nothing enqueued: the stub answers 500 to every request. + suggestions.suggestions(SuggestionScope.NONE, "en"); + awaitIdle(); + suggestions.suggestions(SuggestionScope.NONE, "en"); + awaitIdle(); + int insideTheWindow = MODEL.bodies().size(); + + Thread.sleep(RETRY_AFTER.plusMillis(300).toMillis()); + List after = suggestions.suggestions(SuggestionScope.NONE, "en"); + awaitIdle(); + int afterTheWindow = MODEL.bodies().size(); + suggestions.suggestions(SuggestionScope.NONE, "en"); + awaitIdle(); + + assertThat(insideTheWindow).isEqualTo(1); + assertThat(afterTheWindow).isEqualTo(2); + assertThat(MODEL.bodies()).hasSize(2); + assertThat(after).extracting(Suggestion::text).containsExactlyElementsOf(CATALOG_EN); + assertThat(storedRows()).isZero(); + } + + @Test + void anUnbuildableChatModelIsTriedOncePerRetryWindowNotOncePerOpen(CapturedOutput output) + throws IOException, InterruptedException { + // Named, but with nowhere to send a request: the model cannot be built. + configure(ORIGINAL.replaceAll("(?m)^ base_url: .*$", " base_url: \"\"")); + + for (int open = 0; open < 3; open++) { + assertThat(suggestions.suggestions(SuggestionScope.NONE, "en")) + .extracting(Suggestion::text) + .containsExactlyElementsOf(CATALOG_EN); + awaitIdle(); + } + long insideTheWindow = unbuildable(output); + Thread.sleep(RETRY_AFTER.plusMillis(300).toMillis()); + suggestions.suggestions(SuggestionScope.NONE, "en"); + awaitIdle(); + + assertThat(insideTheWindow).isEqualTo(1); + assertThat(unbuildable(output)).isEqualTo(2); + assertThat(MODEL.bodies()).isEmpty(); + assertThat(chatCalls()).isZero(); + } + + @Test + void aBurstOfStatusChangesInsideTheQuietWindowCostsOneCallForTheNewestSnapshot() throws InterruptedException { + suggestions.suggestions(SuggestionScope.NONE, "en"); + long application = applications.open(offer(), ApplicationStatus.NEW); + awaitIdle(); + MODEL.reset(); + MODEL.enqueue(ModelStub.text(Duration.ZERO, phrased("1"))); + MODEL.enqueue(ModelStub.text(Duration.ZERO, phrased("1"))); + MODEL.enqueue(ModelStub.text(Duration.ZERO, phrased("1"))); + + burst(application, 6, () -> {}); + awaitIdle(); + int afterTheBurst = MODEL.bodies().size(); + List shown = suggestions.suggestions(SuggestionScope.NONE, "en"); + awaitIdle(); + int afterTheOpen = MODEL.bodies().size(); + + move(application, ApplicationStatus.SHORTLISTED); + awaitIdle(); + move(application, ApplicationStatus.NEW); + awaitIdle(); + + assertThat(afterTheBurst).isEqualTo(1); + assertThat(afterTheOpen) + .as("the burst phrased the newest snapshot, so the open is a hit") + .isEqualTo(1); + assertThat(shown).extracting(Suggestion::text).contains("How are the 1 open applications doing?"); + assertThat(MODEL.bodies()) + .as("two changes a quiet window apart cost two calls") + .hasSize(3); + } + + @Test + void reopeningTheChatAfterEveryMoveCostsOneCallPerPauseNotOnePerMove() throws InterruptedException { + suggestions.suggestions(SuggestionScope.NONE, "en"); + long application = applications.open(offer(), ApplicationStatus.NEW); + awaitIdle(); + MODEL.reset(); + MODEL.enqueue(ModelStub.text(Duration.ZERO, phrased("1"))); + MODEL.enqueue(ModelStub.text(Duration.ZERO, phrased("1"))); + + // The docked panel asks again after each move: every open is a new snapshot, and a miss. + burst(application, 5, () -> suggestions.suggestions(SuggestionScope.NONE, "en")); + awaitIdle(); + + assertThat(MODEL.bodies()) + .as("five moves and five missed opens inside one quiet window") + .hasSize(1); + } + + @Test + void aChangeIsPreparedWhileTheSchedulerIsBusyWithAPass() throws InterruptedException { + suggestions.suggestions(SuggestionScope.NONE, "en"); + long application = applications.open(offer(), ApplicationStatus.NEW); + awaitIdle(); + MODEL.reset(); + MODEL.enqueue(ModelStub.text(Duration.ZERO, phrased("1"))); + + // A scheduled ingest pass holds Boot's one scheduler thread for as long as it runs. + CountDownLatch passRunning = new CountDownLatch(1); + CountDownLatch passDone = new CountDownLatch(1); + scheduler.schedule( + () -> { + passRunning.countDown(); + try { + passDone.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }, + Instant.now()); + try { + assertThat(passRunning.await(10, TimeUnit.SECONDS)).isTrue(); + move(application, ApplicationStatus.SHORTLISTED); + // Polled rather than slept: a loaded runner may take longer than any fixed pause, and + // only the pass's end may not be what lets the window pass. + long deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos(); + while (MODEL.bodies().isEmpty() && System.nanoTime() < deadline) { + Thread.sleep(20); + } + + assertThat(passDone.getCount()) + .as("the pass still holds the scheduler") + .isOne(); + assertThat(MODEL.bodies()) + .as("the quiet window passed while the pass still held the scheduler") + .hasSize(1); + } finally { + passDone.countDown(); + } + awaitIdle(); + } + + @Test + void aMissIsPhrasedInItsOwnLanguageWhateverTheNextOpenAsks() { + // The worker is held on a pinned miss while a German and then an English open miss. + MODEL.enqueue(ModelStub.text(Duration.ofMillis(700), phrased("0"))); + MODEL.enqueue(ModelStub.text(Duration.ZERO, phrased("0"))); + MODEL.enqueue(ModelStub.text(Duration.ZERO, phrased("0"))); + suggestions.suggestions(new SuggestionScope(42L, List.of()), "en"); + awaitFirstRequest(); + + suggestions.suggestions(SuggestionScope.NONE, "de"); + suggestions.suggestions(SuggestionScope.NONE, "en"); + awaitIdle(); + + assertThat(MODEL.bodies()).hasSize(3); + assertThat(MODEL.bodies().get(1)) + .as("the German miss, though the last open asked in English") + .contains("German"); + assertThat(MODEL.bodies().get(2)).doesNotContain("German"); + } + + private static long unbuildable(CapturedOutput output) { + return output.getOut() + .lines() + .filter(line -> line.contains("is named but could not be built")) + .count(); + } + private int storedRows() { // Only this test's snapshot: earlier tests of the class stored theirs under older run ids. return jdbc.queryForObject( @@ -393,6 +587,32 @@ private void move(long application, ApplicationStatus to) { applications.update(application, new ApplicationUpdate(to, null, null, null, null, null)); } + /** + * {@code changes} status moves a sixth of the quiet window apart, so together they span about one + * window and only a window restarted by each change holds them together; {@code afterEach} runs + * after every move. A runner so loaded that two committed moves came most of a window apart + * cannot show a burst at all, so the case is skipped then rather than failed. + */ + private void burst(long application, int changes, Runnable afterEach) throws InterruptedException { + ApplicationStatus status = ApplicationStatus.NEW; + long previous = 0; + long widest = 0; + for (int change = 0; change < changes; change++) { + status = status == ApplicationStatus.NEW ? ApplicationStatus.SHORTLISTED : ApplicationStatus.NEW; + move(application, status); + long committed = System.nanoTime(); + if (change > 0) { + widest = Math.max(widest, committed - previous); + } + previous = committed; + afterEach.run(); + Thread.sleep(QUIET.toMillis() / 6); + } + assumeThat(Duration.ofNanos(widest)) + .as("the runner kept two moves of the burst well inside one quiet window") + .isLessThan(QUIET.multipliedBy(4).dividedBy(5)); + } + /** The language the warm-up phrased in: whatever the last open of an earlier test asked. */ private String lastPhrasedLanguage() { return MODEL.bodies().getLast().contains("German") ? "de" : "en"; diff --git a/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionWarmupLifecycleTest.java b/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionWarmupLifecycleTest.java new file mode 100644 index 0000000..34f08e9 --- /dev/null +++ b/backend/src/test/java/de/codeministry/leadgen/chat/SuggestionWarmupLifecycleTest.java @@ -0,0 +1,218 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright 2026 Marcello Muscara (codeministry) + * + * Licensed under the Apache License, Version 2.0. You may obtain a copy of the + * License at http://www.apache.org/licenses/LICENSE-2.0 + */ +package de.codeministry.leadgen.chat; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatNoException; + +import de.codeministry.leadgen.Databases; +import de.codeministry.leadgen.analytics.PipelineRunRecorder; +import de.codeministry.leadgen.application.ApplicationService; +import de.codeministry.leadgen.application.ApplicationStatus; +import de.codeministry.leadgen.application.ApplicationUpdate; +import de.codeministry.leadgen.chat.suggest.SuggestionScope; +import de.codeministry.leadgen.chat.suggest.SuggestionService; +import de.codeministry.leadgen.chat.suggest.SuggestionWarmup; +import de.codeministry.leadgen.event.DataChanged; +import java.nio.file.Path; +import java.sql.Timestamp; +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.MethodOrderer; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestMethodOrder; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.postgresql.PostgreSQLContainer; + +/** + * Spec 028: the warm-up's two ends, which no shared context can hold. + * + *

Before the first open since start, a data change prepares nothing: no model request and no + * call from the chat's day, and the first open arms it (ISC-536). That needs a context no case has + * opened the chat in yet, so it runs first. + * + *

After the warm-up was shut down, a data change — a run finishing while the context closes — is + * dropped quietly. Before, the warm-up's own executor refused the task, the refusal reached the + * writer through the synchronous listener, and the writer logged a run it had recorded as one it + * had not (ISC-540). That shuts the warm-up down, so it runs last and the context is discarded. + */ +@SpringBootTest +@Testcontainers +@ExtendWith(OutputCaptureExtension.class) +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +@TestMethodOrder(MethodOrderer.OrderAnnotation.class) +class SuggestionWarmupLifecycleTest { + + private static final ModelStub MODEL = ModelStub.start(); + + private static final Path STUBBED = MODEL.configuration().resolve("pipeline.yaml"); + + private static final Duration QUIET = Duration.ofMillis(300); + + @Container + @ServiceConnection + static final PostgreSQLContainer POSTGRES = Databases.postgres(); + + @DynamicPropertySource + static void configuration(DynamicPropertyRegistry registry) { + registry.add("leadgen.config-dir", () -> STUBBED.getParent().toString()); + registry.add("leadgen.chat.suggestion-quiet-window", () -> QUIET.toString()); + } + + @Autowired + private SuggestionWarmup warmup; + + @Autowired + private PipelineRunRecorder recorder; + + @Autowired + private ApplicationEventPublisher events; + + @Autowired + private JdbcTemplate jdbc; + + @Autowired + private SuggestionService suggestions; + + @Autowired + private ApplicationService applications; + + @AfterAll + static void stop() { + MODEL.close(); + } + + @Test + @Order(1) + void beforeTheFirstOpenADataChangePreparesNothingAndTheFirstOpenArmsIt() throws InterruptedException { + MODEL.enqueue(ModelStub.text(Duration.ZERO, PHRASED)); + MODEL.enqueue(ModelStub.text(Duration.ZERO, PHRASED)); + awaitingRun(); + recorder.complete(0, 0, false); + long application = applications.open(offer(), ApplicationStatus.NEW); + for (ApplicationStatus next : List.of( + ApplicationStatus.SHORTLISTED, + ApplicationStatus.NEW, + ApplicationStatus.SHORTLISTED, + ApplicationStatus.NEW)) { + applications.update(application, new ApplicationUpdate(next, null, null, null, null, null)); + } + Thread.sleep(QUIET.multipliedBy(3).toMillis()); + awaitIdle(); + int beforeAnyOpen = MODEL.bodies().size(); + int chatCallsBeforeAnyOpen = chatCalls(); + + // A pinned open misses as before, is phrased in the background, and arms the warm-up. + suggestions.suggestions(new SuggestionScope(42L, List.of()), "en"); + awaitIdle(); + int afterThePinnedOpen = MODEL.bodies().size(); + applications.update( + application, new ApplicationUpdate(ApplicationStatus.SHORTLISTED, null, null, null, null, null)); + awaitIdle(); + + assertThat(beforeAnyOpen).isZero(); + assertThat(chatCallsBeforeAnyOpen).isZero(); + assertThat(afterThePinnedOpen).isEqualTo(1); + assertThat(MODEL.bodies()) + .as("armed: a change now prepares the next open") + .hasSize(2); + } + + @Test + @Order(2) + void aDataChangeAfterShutdownIsDroppedWithoutReachingTheWriter(CapturedOutput output) { + MODEL.reset(); + warmup.destroy(); + Instant now = Instant.now(); + jdbc.update(""" + INSERT INTO pipeline_run ( + started_at, finished_at, ruleset_version, score_model, status, + documents, extracted, written, merged, + filter_considered, filter_passed, + enrich_considered, enriched, incomplete, from_cache, requests, + score_considered, scored, unscored, shortlisted, review, submitted, + packaged, digest_written) + VALUES (?, ?, '1', 'm', 'AWAITING_BATCH', 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, true) + """, Timestamp.from(now.minusSeconds(60)), Timestamp.from(now)); + + assertThatNoException().isThrownBy(() -> events.publishEvent(new DataChanged("test, after shutdown"))); + recorder.complete(0, 0, false); + + assertThat(jdbc.queryForObject("SELECT status FROM pipeline_run ORDER BY id DESC LIMIT 1", String.class)) + .isEqualTo("COMPLETE"); + assertThat(output.getOut() + output.getErr()) + .doesNotContain("was not completed in the history") + .doesNotContain("The run was not recorded"); + assertThat(warmup.idle()).isTrue(); + assertThat(MODEL.bodies()).isEmpty(); + } + + private static final String PHRASED = "{\"shortlist\":\"Which of the 0 shortlisted offers fits?\"," + + "\"openApplications\":\"How are the 0 open applications doing?\"}"; + + private void awaitingRun() { + Instant now = Instant.now(); + jdbc.update(""" + INSERT INTO pipeline_run ( + started_at, finished_at, ruleset_version, score_model, status, + documents, extracted, written, merged, + filter_considered, filter_passed, + enrich_considered, enriched, incomplete, from_cache, requests, + score_considered, scored, unscored, shortlisted, review, submitted, + packaged, digest_written) + VALUES (?, ?, '1', 'm', 'AWAITING_BATCH', 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, true) + """, Timestamp.from(now.minusSeconds(60)), Timestamp.from(now)); + } + + private long offer() { + long source = jdbc.queryForObject( + "INSERT INTO source (name, kind) VALUES ('lifecycle', 'file') RETURNING id", Long.class); + return jdbc.queryForObject(""" + INSERT INTO offer (source_id, external_id, title, url, fingerprint, status, score_band, + score_value, tags, ingested_at) + VALUES (?, 'lifecycle', 'Offer lifecycle', 'https://example.invalid/lifecycle', 'lifecycle', + 'PASSED', 'SHORTLISTED', 95, '{}'::text[], now()) + RETURNING id + """, Long.class, source); + } + + private int chatCalls() { + return jdbc.queryForObject("SELECT coalesce(sum(calls), 0) FROM chat_call_budget", Integer.class); + } + + private void awaitIdle() { + long deadline = System.nanoTime() + Duration.ofSeconds(30).toNanos(); + while (!warmup.idle()) { + if (System.nanoTime() > deadline) { + throw new AssertionError("the suggestion warm-up did not finish within 30 s"); + } + try { + Thread.sleep(20); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + } +} diff --git a/backend/src/test/java/de/codeministry/leadgen/chat/suggest/PhrasingWithoutAContextTest.java b/backend/src/test/java/de/codeministry/leadgen/chat/suggest/PhrasingWithoutAContextTest.java new file mode 100644 index 0000000..0e686ab --- /dev/null +++ b/backend/src/test/java/de/codeministry/leadgen/chat/suggest/PhrasingWithoutAContextTest.java @@ -0,0 +1,113 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright 2026 Marcello Muscara (codeministry) + * + * Licensed under the Apache License, Version 2.0. You may obtain a copy of the + * License at http://www.apache.org/licenses/LICENSE-2.0 + */ +package de.codeministry.leadgen.chat.suggest; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import de.codeministry.leadgen.chat.ChatBudget; +import de.codeministry.leadgen.chat.ChatCapability; +import de.codeministry.leadgen.config.ConfigProperties; +import de.codeministry.leadgen.config.ConfigRegistry; +import de.codeministry.leadgen.llm.ChatModels; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.IntStream; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.model.ChatModel; +import org.springframework.ai.chat.model.ChatResponse; +import org.springframework.ai.chat.model.Generation; +import org.springframework.ai.chat.prompt.Prompt; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import reactor.core.publisher.Flux; + +/** + * {@link Phrasing} without a context: what happens when one of its maps reaches + * {@link Phrasing#CACHE_LIMIT}, and what the follow-ups' path says about a model it cannot build. + */ +@ExtendWith(OutputCaptureExtension.class) +class PhrasingWithoutAContextTest { + + private static final Map SENTENCES = Map.of("open", "1 application is open. Where does it stand?"); + + private final ChatCapability capability = mock(ChatCapability.class); + private final ChatModels chatModels = mock(ChatModels.class); + private final ConfigRegistry config = mock(ConfigRegistry.class, RETURNS_DEEP_STUBS); + private final ChatBudget budget = mock(ChatBudget.class); + private final PhrasingStore store = mock(PhrasingStore.class); + private final ConfigProperties properties = mock(ConfigProperties.class, RETURNS_DEEP_STUBS); + + private Phrasing phrasing; + + @BeforeEach + void wire() { + when(capability.model()).thenReturn(Optional.of("model")); + // No llm block: the call falls back to its default timeout. + when(config.snapshot().application().llm()).thenReturn(null); + when(properties.chat().suggestionRetryAfter()).thenReturn(Duration.ofMinutes(5)); + phrasing = new Phrasing(capability, chatModels, config, budget, store, properties); + } + + @Test + void aFullSkipListMakesRoomForOneKeyAndKeepsEveryOtherWindow() { + // The model is named but cannot be built: every attempt is skipped for the retry window. + when(chatModels.of(any(), any())).thenReturn(Optional.empty()); + List keys = IntStream.rangeClosed(0, Phrasing.CACHE_LIMIT) + .mapToObj(i -> phrasing.key("suggestions " + i, SENTENCES, "en").orElseThrow()) + .toList(); + + for (int i = 0; i <= Phrasing.CACHE_LIMIT; i++) { + phrasing.attempt("suggestions " + i, SENTENCES, "en"); + } + + assertThat(keys.stream().filter(phrasing::skipped)) + .as("one more skip than the bound drops one window, not all of them") + .hasSize(Phrasing.CACHE_LIMIT); + assertThat(phrasing.skipped(keys.getLast())).isTrue(); + } + + @Test + void followUpsNeverPushASuggestionPhrasingOutOfMemory() { + String suggestion = phrasing.key("suggestions 1", SENTENCES, "en").orElseThrow(); + when(store.find(suggestion)).thenReturn(Optional.of(SENTENCES)); + assertThat(phrasing.stored(suggestion)).isPresent(); + + ChatModel model = mock(ChatModel.class); + when(model.stream(any(Prompt.class))) + .thenAnswer(call -> Flux.just(new ChatResponse(List.of(new Generation(new AssistantMessage("{}")))))); + when(chatModels.of(any(), any())).thenReturn(Optional.of(model)); + when(budget.take()).thenReturn(true); + for (int turn = 0; turn <= Phrasing.CACHE_LIMIT; turn++) { + phrasing.phrase("followups " + turn, SENTENCES, "en"); + } + + assertThat(phrasing.stored(suggestion)).isPresent(); + verify(store, times(1).description("the suggestion is still in memory, not read from the table again")) + .find(suggestion); + } + + @Test + void aFollowUpWhoseModelCannotBeBuiltSaysSoAndShowsTheCatalog(CapturedOutput output) { + when(chatModels.of(any(), any())).thenReturn(Optional.empty()); + + assertThat(phrasing.phrase("followups 1", SENTENCES, "en")).isEqualTo(SENTENCES); + assertThat(output.getOut()).contains("is named but could not be built; the follow-ups show"); + } +} diff --git a/backend/src/test/java/de/codeministry/leadgen/config/ConfigPropertiesChatTest.java b/backend/src/test/java/de/codeministry/leadgen/config/ConfigPropertiesChatTest.java index b695d61..7d25c49 100644 --- a/backend/src/test/java/de/codeministry/leadgen/config/ConfigPropertiesChatTest.java +++ b/backend/src/test/java/de/codeministry/leadgen/config/ConfigPropertiesChatTest.java @@ -30,6 +30,8 @@ void theBoundDefaultsAreTheHandBuiltOnes() { assertThat(bound).isEqualTo(new ConfigProperties("config")); assertThat(bound.chat().turnTimeout()).isPositive(); assertThat(bound.chat().heartbeat()).isLessThan(bound.chat().turnTimeout()); + assertThat(bound.chat().suggestionRetryAfter()).isEqualTo(Duration.ofMinutes(5)); + assertThat(bound.chat().suggestionQuietWindow()).isEqualTo(Duration.ofSeconds(30)); } @Test @@ -37,10 +39,14 @@ void aSetKeyOverridesItsDefault() { ConfigProperties bound = bind(Map.of( "leadgen.config-dir", "config", "leadgen.chat.turn-timeout", "PT2S", - "leadgen.chat.heartbeat", "PT0.15S")); + "leadgen.chat.heartbeat", "PT0.15S", + "leadgen.chat.suggestion-retry-after", "PT2S", + "leadgen.chat.suggestion-quiet-window", "PT0.3S")); assertThat(bound.chat().turnTimeout()).isEqualTo(Duration.ofSeconds(2)); assertThat(bound.chat().heartbeat()).isEqualTo(Duration.ofMillis(150)); + assertThat(bound.chat().suggestionRetryAfter()).isEqualTo(Duration.ofSeconds(2)); + assertThat(bound.chat().suggestionQuietWindow()).isEqualTo(Duration.ofMillis(300)); } private static ConfigProperties bind(Map properties) { diff --git a/docs/DATA-MODEL.md b/docs/DATA-MODEL.md index 72a97d4..b2d9a98 100644 --- a/docs/DATA-MODEL.md +++ b/docs/DATA-MODEL.md @@ -565,16 +565,18 @@ every ended turn from `V40` on. Read by `spend` for `GET /chat/status`: today an per connection and in total, in one `GROUPING SETS` query. No key points at it and nothing deletes from it; emptying it forgets what was spent. -**`chat_suggestion_phrasing`** (`V39`). The empty chat's suggestions and a turn's follow-ups as the -chat model phrased them, one row per phrasing, so a restart costs a table read rather than a model -call (spec 027, ISC-531). `cache_key` is a SHA-256 of the model, the language, the snapshot and the -catalog sentences; `snapshot` (the caller's key text: `suggestions : ` or -`followups `), `language`, `model` and `sentences` are there to read, `phrased` is what is -shown, a catalog sentence wherever the number guard refused the model's. `PhrasingStore.save` inserts -after an accepted call with `ON CONFLICT DO NOTHING` and prunes to the newest 256 rows by -`created_at` (indexed); `find` reads by key when the map in front of it misses, and `newestLanguage` -the language a warm-up uses before the first open since start. No key points at it and it points at -nothing: emptying it costs the next open per key a background call. +**`chat_suggestion_phrasing`** (`V39`). The empty chat's suggestions as the chat model phrased them, +one row per phrasing, so a restart costs a table read rather than a model call (spec 027, ISC-531). +`cache_key` is a SHA-256 of the model, the language, the snapshot and the catalog sentences; +`snapshot` (the caller's key text, `suggestions : `), `language`, `model` +and `sentences` are there to read, `phrased` is what is shown, a catalog sentence wherever the +number guard refused the model's. A turn's follow-ups are phrased through the same call but kept in +memory only (spec 028, ISC-539): they are never asked for again, and rows of theirs pushed the +suggestions out of the bound. Rows an earlier release stored for them (`followups `) age +out; the table comment still names them, since an applied migration is not edited. `PhrasingStore.save` +inserts after an accepted call with `ON CONFLICT DO NOTHING` and prunes to the newest 256 rows by +`created_at` (indexed); `find` reads by key when the map in front of it misses. Nothing reads a +language back from it. No key points at it and it points at nothing: emptying it costs the next open per key a background call. ## 6. Status-like columns and their values diff --git a/docs/decisions/chat.md b/docs/decisions/chat.md index e0e2ade..acdc0f4 100644 --- a/docs/decisions/chat.md +++ b/docs/decisions/chat.md @@ -569,22 +569,43 @@ a new key; the map clears past 256 entries, because an old snapshot is never ask first open after every run, every status change and every restart showed skeleton rows for as long as a local model took, up to the timeout, to reword sentences the rules had already found. Now `GET …/suggestions` answers what is stored for its key, or the catalog's sentences, and hands -a miss to `SuggestionWarmup`: one worker thread, one model call at a time. The unpinned -suggestions are a single slot — asking for them while a call runs marks them due once more, and -that one more is computed from the data as it is then, so five status changes during a slow call -cost one further call for the newest snapshot. A pinned context carries its own sentences and -waits in a de-duplicated set of at most 16. A key whose call failed is not asked again until the -data, the model or the language moves it, so a model that is down costs one request per key and -not one per open. A sentence on screen is never swapped: the panel asks once per target, and the +a miss to `SuggestionWarmup`, one model call at a time. A miss, pinned or not, hands over the job +its request built — snapshot, sentences, language — and waits in a set de-duplicated by key, of at +most 16, so a German miss followed by an English open is phrased once in each language, and the +worker never reads the data a second time (spec 028, ISC-538). A data change marks one slot due +instead, computed from the data as it is when the worker gets to it, so five status changes +during a slow call cost one further call for the newest snapshot. A key whose call failed, or +whose named model cannot be built, is left alone for `leadgen.chat.suggestion-retry-after` (five +minutes) and asked once more after it, so a model that is down costs one request per key and +window, and a cold start that timed out is over by the next try (ISC-535). One call at a time is +the warm-up's own `running` flag, not a pool of one: the work runs on Boot's +`applicationTaskExecutor` like every other background task, and after shutdown a late trigger is +dropped rather than thrown back at the writer (ISC-540). A sentence on screen is never swapped: the panel asks once per target, and the phrasing shows from the next open on. **The warm-up follows the data, not the open.** `PipelineRunRecorder` publishes -`SuggestionDataChanged` at a run's end (not its start, when the run id already moved but the +`event.DataChanged` at a run's end (not its start, when the run id already moved but the counts have not settled), and so does every writer of an `application_event`: `ApplicationService` and the restore in `ArchiveService`. A new writer of that table publishes it too. The listener runs -after the commit and only marks the unpinned slot due, so neither a run nor a status write waits on -the model or fails with it. It prepares the language the last open asked in, and before the first -open since start the newest stored phrasing's. Polling `snapshot()` would also catch direct SQL +after the commit and only schedules, so neither a run nor a status write waits on the model or +fails with it. It prepares the language the last open asked in. The event lives in `event`, a +package no consumer owns, because the pipeline must not import the chat it promises can be removed +(ISC-422); `ChatIsolationTest` holds both halves, every `chat_*` table the migrations create and +every import of the chat's package (spec 028, ISC-541). + +**Paid for only when the chat is used, and once per quiet spell (spec 028).** Each warm-up is a +call from the chat's day, and triaging a hundred applications one by one spent half of a 200-call +day on a chat nobody opened. So a data change prepares nothing until the chat has been opened +since start (ISC-536), and after that the warm-up starts only once +`leadgen.chat.suggestion-quiet-window` (30 s) has passed without a further change, so a burst costs +one call for the newest snapshot (ISC-537). The window is timed by `CompletableFuture.delayedExecutor`, +not by Boot's `taskScheduler`: that scheduler has one thread, a scheduled ingest pass holds it for +as long as the pass runs, and a window queued behind it passed only when the pass ended. Either alone +falls short: the quiet window still pays per spell for a chat nobody opens, the gate still pays per +change once it has been opened. The window and the retry are `ConfigProperties.Chat` durations with +defaults, not `.env` values, so a test can shorten them; no deployment needs to set them. The +price: the first open after a start misses once, and an open inside the quiet window after a +change is a miss answered with the catalog. Polling `snapshot()` would also catch direct SQL against the live instance, but leaves a window of one interval in which the first open is a miss; `LISTEN/NOTIFY` closes that window at the price of a trigger in a migration and a listener connection that must survive reconnects. A write that publishes nothing is still covered: its @@ -593,7 +614,9 @@ next open misses, answers the catalog at once and phrases in the background. **The phrasing outlives the process.** `chat_suggestion_phrasing` (`V39`) holds each phrasing under a SHA-256 of the key, behind the map, so a restart costs a table read rather than a model call; two instances phrasing one key keep the first row, and the table keeps the newest 256. It is a cache: -dropping it costs the next open per key a background call. Postgres rather than Redis or Valkey, +dropping it costs the next open per key a background call. Follow-ups are kept in memory only: a +finished turn's follow-ups are never asked for again, and their rows used to push the suggestions +out of the bound (ISC-539). Postgres rather than Redis or Valkey, because a second store for a handful of rows is a service in compose, the chart and every deployment, and the tool already runs and tests one.