Compare commits

..
Author SHA1 Message Date
james.maesandClaude Opus 5.5 9fc0eed499 fix(runtime): shutdown hook holds the runtime weakly
Every AssistantRuntime registered a JVM shutdown hook that held the
runtime, and so its whole QInstance, until JVM exit. A host test suite
that builds an instance per test leaked every instance (kof22's suite ran
out of heap). The hook now holds a weak reference.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 15:53:47 -05:00
james.maes 23779790ab Merge pull request #55 from QRun-IO/fix/help-bundle-version
fix(help): help bundle version comes from the build, not a hardcoded SNAPSHOT
2026-10-14 22:37:20 -05:00
james.maes bbee0d1017 Merge pull request #45 from QRun-IO/feature/health-checks
feat(health): register assistant provider and model checks with core central health
2026-10-14 22:08:14 -05:00
james.maesandClaude Opus 5.5 079e8d39e2 fix(help): the help bundle carries the artifact version (Maven-filtered) instead of a hardcoded SNAPSHOT
On release branches the bundle's version and the producer's differed, so the
bundle test that asserts they match failed, and released bundles recorded
SNAPSHOT as their provenance. help/*.yaml is now filtered:
version: ${project.version}.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 22:03:49 -05:00
james.maes dd5edf84d8 Merge remote-tracking branch 'origin/develop' into feature/health-checks
# Conflicts:
#	CHANGELOG.md
#	pom.xml
2026-10-14 22:02:13 -05:00
james.maes ef933d6bea Merge pull request #54 from QRun-IO/feature/jobs-tray
feat(jobs): assistant answers in the Jobs tray (core JobTracking, QQQ RC.8)
2026-10-14 21:25:22 -05:00
james.maesandClaude Opus 5.5 385b61019b docs: Jobs tray tracking (QQQ RC.8) in CHANGELOG and UPGRADE
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 21:20:46 -05:00
james.maesandClaude Opus 5.5 a7184d3be9 test(jobs): tracked progress and completion, Stop from the tray, failure without transcript text, no runtime, isolated phase state
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 21:20:45 -05:00
james.maesandClaude Opus 5.5 2a3691c0ea feat(jobs): assistant answers in the Jobs tray through core JobTracking
JobTrackingRunTracker is the default AssistantRunTracker and replaces the
qqq#1210 stand-in: each answer is a tracking job (processName
assistant.turn, label "Assistant answer", linked to its conversation)
with thinking / tool / writing phases and the tool's label as the
message. The tray's Stop stops the turn; the turn's finaliser finishes the
job COMPLETED, CANCELLED or FAILED with fixed messages (never prompt or
answer text). No jobs runtime on the pod: answers are simply not tracked.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 21:20:44 -05:00
james.maesandClaude Opus 5.5 b6982eff9f build: QQQ 4.1.0-RC.8 (core job tracking for the Jobs tray)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 21:20:44 -05:00
james.maesandClaude Opus 5.5 2a7bf77c6d Merge origin/develop into feature/health-checks; pin QQQ 4.1.0-RC.8
RC.8 ships the central health kernel (QRun-IO/qqq#1207).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 21:18:21 -05:00
james.maes 1012b4b497 Merge pull request #53 from QRun-IO/feature/help-bundle
feat(help): qbit-assistant help bundle via QQQ help bundles, with coverage gate
2026-10-14 20:29:39 -05:00
james.maesandClaude Opus 5.5 76f9b286c6 feat(help): ship the qbit-assistant help bundle through QQQ help bundles, with coverage gate
Host-neutral help (deployment-specific facts removed), registered with
HelpBundles, gated by HelpBundleAssertions. QQQ 4.1.0-RC.7.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 20:16:55 -05:00
james.maes 9b4708fafc Merge pull request #44 from QRun-IO/feature/transcript-redaction
feat: no raw PII/PHI in stored transcripts (secured data §15, QQQ RC.7)
2026-10-14 19:03:20 -05:00
james.maesandClaude Opus 5.5 02215634bf docs: assistant health checks
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 13:39:46 -05:00
james.maesandClaude Opus 5.5 303cf9a148 refactor(assistant): home shows the central health status instead of its own provider health view
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 13:39:46 -05:00
james.maesandClaude Opus 5.5 969a7b9101 feat(health): register provider and model checks; breaker reuses the central result
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 13:39:45 -05:00
james.maesandClaude Opus 5.5 8e47d9a6f5 feat(health): deadline-aware connection test with shared cost-cap accounting
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 13:39:44 -05:00
james.maesandClaude Opus 5.5 09787bfe22 fix: set forwarded UI API headers instead of appending (QQQ 4.1 develop compatibility)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 13:39:43 -05:00
28 changed files with 3477 additions and 356 deletions

No files matched your search

+25
View File
@@ -1,5 +1,30 @@
# Changelog
## Unreleased - Central health
### Added
- Dynamic `assistant.provider.<id>` and `assistant.model.<id>` dependency checks in component `assistant`, category AI, reusing provider diagnostics and minimal connection calls with deadlines, cost-cap/kill-switch skips and TEST usage accounting.
- LOCAL breaker reuse of central provider outcomes younger than 60 seconds, with its existing lazy probe as fallback.
- Generic Assistant health status card linking to filtered System Health; registration, status, accounting, deadline and sentinel-secret tests.
- QQQ 4.1 compatibility for the header shim: setting extra headers avoids duplicates now that core forwards them too.
### Removed
- The home's independent answering/last-test health indicators and per-provider health/down-until display. Central health owns status and history. Manual Test connection actions, `lastTest*` fields and the circuit breaker remain.
## Unreleased - Jobs tray
- Track assistant answers in the core Jobs tray through `JobTracking`, with thinking, tool and writing progress and Stop support.
- Require QQQ 4.1.0-RC.8 and the host's durable jobs runtime and jobs tables; answers remain untracked when the runtime is absent.
- Default to `JobTrackingRunTracker`; opt out with `withRunTracker(AssistantRunTracker.NOOP)`.
## Unreleased - In-app help
- Ship the host-neutral Assistant help bundle through QQQ's native registry, replacing the temporary loader.
- Cover Assistant tables, fields, actions, widgets and selectors; retain live metadata references and add a bundle coverage gate.
- Document HOST bundle overrides and remove deployment-specific provider, classification, retention and backup guidance.
## Unreleased - Transcript redaction
- Tokenize classified PII/PHI in stored messages, tool inputs/results, replay events, pending actions and derived summaries using QQQ secured-data tokens. Preserve eligible model requests and live SSE.
+49 -5
View File
@@ -220,12 +220,41 @@ Admins set up providers on the **Providers** screen (`assistant_provider`), a st
**Test connection** is a record action on a provider (select one or more). It is a standard streamed ETL: Review lists what will be called and anything wrong with the row (missing key, disallowed variable or URL) before any call. Result has one line per provider: `connected to claude-sonnet-5-5 in 812 ms`, or the error with its kind and HTTP status. It makes one short live call (a 1-sentence prompt, at most 32 output tokens, effort low), saves the result on the row, and records the call in `assistant_usage` with purpose `TEST`.
## Health checks
The producer registers dynamic central checks in component `assistant`, category `AI`:
| Name | Kind | Interval | Timeout |
| --- | --- | --- | --- |
| `assistant.provider.<id>` | DEPENDENCY | 300s; 60s for LOCAL providers | 10s |
| `assistant.model.<id>` | DEPENDENCY | 21600s (6 hours) | 30s |
Suppliers include enabled provider/model rows and refresh with the central registry.
Provider checks reuse `AssistantProviderType.checkHealth`. Unsupported probes use
circuit-breaker state only when less than ten minutes old, otherwise return UNKNOWN.
Model checks reuse `ConnectionTest` (embeddings for embedding models), respect the
provider's monthly cost cap and the kill switch, and leave manual `lastTest*` fields
unchanged. A capped check returns UNKNOWN, `skipped: cost cap`, even on demand.
The automated runner records usage under its own owner key. Provider caps use the
existing shared accounting projection without access to other users' usage records.
Usage purpose is a string, so health diagnostics reuse `TEST` (not a new `HEALTH` enum).
All checks honor their deadlines; messages and details omit provider error text,
credentials, endpoints, prompts and completions.
The LOCAL circuit breaker consults central results less than 60 seconds old before
its lazy probe. UP/DEGRADED avoids a duplicate probe, DOWN rejects the call, and
UNKNOWN or stale/missing results retain the existing probe fallback. The breaker
and manual Test connection actions remain available.
This integration uses the core health SPI released in QQQ 4.1.0-RC.8.
The POM pins that release; verify with `./mvnw -o verify`.
## The app home
The **Assistant** app's home is the one place for the assistant's health, usage and cost (no other dashboard repeats it). All of it comes from the tables below, read as the viewing admin; an empty table gives an empty state, never sample data.
The **Assistant** app's home shows configuration, usage and cost read as the viewing admin. A generic status card summarizes the central runner's latest results and links to System Health filtered to `assistant`; it shows UNKNOWN before results are available. Health history and check lists live in System Health.
- **Health:** is it answering (kill switch, default provider, key present, last test), the 7-day error rate of model calls, cost and requests this month against the instance's monthly limit.
- **Providers:** one row per provider: type, model, masked key status, last test, cost this month against its cap, limits, and any problem with its settings.
- **Configuration and usage:** default provider, key status, the 7-day error rate of model calls, cost and requests this month against the instance's monthly limit.
- **Providers:** one row per provider: type, model, masked key status, cost this month against its cap, limits, and any problem with its settings.
- **Usage, last 30 days:** bars of model requests per day, totals (conversations, requests, tokens in and out, cost) and the last 7 days in words.
- **Costs this month:** month to date against the policy limit and each provider cap, then by model. Cost is tokens times the model's price from **Model prices**.
- **By user, this month**, **Top tools** (calls, failures, average time) and **Recent failures** (model calls and tool calls, linked).
@@ -644,8 +673,8 @@ never automatically move to cloud. Provider state/signatures are removed when
switching, preserving neutral tool history. Each attempted call is audited as usage.
NETWORK/TIMEOUT marks a provider DOWN for 60 seconds; success marks it UP. Local
health is probed when unknown/stale with a two-second timeout. Provider cards show
health and down-until, failures show fallback totals, and turn records show the
health uses recent central results or a lazy probe with a two-second timeout. The
central health card links to System Health; failures show fallback totals, and turn records show the
route and final destination. See [UPGRADE.md](UPGRADE.md) for the additive schema.
## Scheduled tasks
@@ -725,3 +754,18 @@ remain untouched, as do explicit object and permission-rule denials.
History visibility uses core's `isDenied` read/all-operation check. RC.6 has no
operation argument; the later write-only deny patterns must remain separate
from that check so they do not hide the read tool.
## In-app help
The QBit registers [`help/qbit-assistant.yaml`](qbit-assistant/src/main/resources/help/qbit-assistant.yaml)
with QQQ's help-bundle registry during production. Hosts with help content storage use QQQ's
startup bundle sync; registration is harmless when the host has no `helpContent` table.
To customize the wording, register a separate bundle with `layer: HOST` using
`HelpBundles.register(qInstance, "help/my-application.yaml")`. Entries with the same key and
role override this QBIT bundle. User-edited help remains protected by core's sync.
Keep live `{{...}}` references so labels and metadata values follow the host's configuration.
`AssistantTablesGuardrailTest.everyOwnedTargetHasBundleHelp` checks coverage and live references
against the produced test instance with `HelpBundleAssertions`; it runs in the normal Maven build.
Update the bundle's literal version alongside the artifact version when releasing.
+12 -6
View File
@@ -69,12 +69,18 @@ reattach through the existing events route. Consumers should recognize
`turn.phase` and `message.start.startedAt`. Starts during shutdown return
503 `SHUTTING_DOWN`; clients may retry against a ready replica.
Hosts may wire `AssistantRunTracker` for Jobs tray publication, progress, Stop,
and terminal notification. It is a no-op by default and all callbacks are
isolated from turn execution. Tracking ids are local to the owning runtime;
reaping an already dead owner does not invent another tracking job. Hosts should
associate their durable tracking row with the supplied turn id. The SPI is a
stand-in for QQQ #1210.
`JobTrackingRunTracker` is now the default `AssistantRunTracker`, using core
`JobTracking` in QQQ **4.1.0-RC.8** to publish answers in the Jobs tray. The host
must provide the jobs tables (through qbit-jobs / core `DurableJobsMetaDataProvider`)
and run `DurableJobsRuntime`. Without that runtime, answers complete normally
and are simply not tracked.
Each job is owned by the assistant user, links to its `assistant_conversation`,
and reports thinking, tool and writing phases. Tool progress contains only the
tool label; terminal messages are fixed and never contain prompts or answers.
The tray's Stop cancels the assistant turn. All tracker callbacks remain isolated
from turn execution. Opt out with `withRunTracker(AssistantRunTracker.NOOP)`;
hosts may still supply their own `AssistantRunTracker`.
## Starter defaults, setup card and table help
+4 -3
View File
@@ -202,7 +202,7 @@
</build>
<profiles>
<!-- The assistant needs QQQ 4.1.0-RC.6: the AI tool kernel (qqq#1078), the -->
<!-- The assistant needs QQQ 4.1.0-RC.7: the AI tool kernel (qqq#1078), the -->
<!-- developer panels (qqq#1076) and the UI API extension point, invoker -->
<!-- and credentials (qqq#1128). Until -->
<!-- qbit-build-parent imports the 4.1 BOM, this -->
@@ -215,8 +215,9 @@
<activeByDefault>true</activeByDefault>
</activation>
<properties>
<!-- RC.7 carries qqq#1182 classified-value tokens and log redaction. Keep the PR draft until published. -->
<qqq.snapshot.version>4.1.0-RC.7</qqq.snapshot.version>
<!-- RC.7 provides classified-value tokens, log redaction and native help bundles; -->
<!-- RC.8 provides core job tracking (JobTracking, the Jobs tray). -->
<qqq.snapshot.version>4.1.0-RC.8</qqq.snapshot.version>
</properties>
<dependencyManagement>
<dependencies>
+2
View File
@@ -114,6 +114,7 @@
<directory>src/main/resources</directory>
<filtering>true</filtering>
<includes>
<include>help/*.yaml</include>
<include>qbit-assistant.properties</include>
</includes>
</resource>
@@ -121,6 +122,7 @@
<directory>src/main/resources</directory>
<filtering>false</filtering>
<excludes>
<exclude>help/*.yaml</exclude>
<exclude>qbit-assistant.properties</exclude>
</excludes>
</resource>
@@ -33,6 +33,7 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import com.kingsrook.qbits.assistant.classification.ClassificationSource;
import com.kingsrook.qbits.assistant.classification.ConfiguredClassificationSource;
import com.kingsrook.qbits.assistant.jobs.AssistantRunTracker;
import com.kingsrook.qbits.assistant.jobs.JobTrackingRunTracker;
import com.kingsrook.qbits.assistant.rag.VectorStore;
import com.kingsrook.qbits.assistant.spi.AssistantModelProvider;
import com.kingsrook.qbits.assistant.spi.AssistantProviderType;
@@ -106,7 +107,7 @@ public class AssistantQBitConfig implements QBitConfig
private Integer shutdownGraceSeconds = 8;
private String defaultConcurrencyMode = "SUPERSEDE";
private Boolean registerShutdownHook = true;
private AssistantRunTracker runTracker = AssistantRunTracker.NOOP;
private AssistantRunTracker runTracker = new JobTrackingRunTracker();
private Integer maxTurnSeconds = 300;
private Integer maxUserMessageChars = 8000;
@@ -23,10 +23,11 @@ import java.util.List;
import java.util.Properties;
import com.kingsrook.qbits.assistant.api.AssistantRoutes;
import com.kingsrook.qbits.assistant.developer.AssistantDeveloperPanels;
import com.kingsrook.qbits.assistant.health.AssistantHealthChecks;
import com.kingsrook.qbits.assistant.health.AssistantHealthWidget;
import com.kingsrook.qbits.assistant.home.AssistantAccessWidget;
import com.kingsrook.qbits.assistant.home.AssistantHome;
import com.kingsrook.qbits.assistant.metadata.AssistantChannelWall;
import com.kingsrook.qbits.assistant.metadata.AssistantHelp;
import com.kingsrook.qbits.assistant.metadata.AssistantInstanceMetaData;
import com.kingsrook.qbits.assistant.metadata.AssistantMetaDataBuilder;
import com.kingsrook.qbits.assistant.metadata.AssistantRecordScreens;
@@ -43,6 +44,7 @@ import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerMultiOutput;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.help.bundles.HelpBundles;
import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.permissions.ClientChannelsMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
@@ -200,11 +202,13 @@ public class AssistantQBitProducer implements QBitMetaDataProducer<AssistantQBit
qInstance.addProcess(process);
}
qInstance.addWidget(AssistantHome.widget());
qInstance.addWidget(AssistantHealthWidget.widget());
qInstance.addWidget(AssistantAccessWidget.widget());
QAppMetaData app = builder.buildAdminApp(tables);
qInstance.addApp(app);
AssistantHelp.apply(qInstance);
// Registration is metadata-only; core startup sync guards hosts without helpContent.
HelpBundles.register(qInstance, "help/qbit-assistant.yaml");
AssistantChannelWall.register(qInstance);
if(Boolean.TRUE.equals(qBitConfig.getRecordHistory()))
@@ -216,6 +220,7 @@ public class AssistantQBitProducer implements QBitMetaDataProducer<AssistantQBit
ClientChannelsMetaData.ofOrWithNew(qInstance).withDeniedPattern("assistant", "audit*");
}
postProduceActions(new MetaDataProducerMultiOutput(), qInstance);
AssistantHealthChecks.register(qInstance);
return (new MetaDataProducerMultiOutput());
}
@@ -17,6 +17,7 @@
package com.kingsrook.qbits.assistant;
import java.lang.ref.WeakReference;
import java.net.InetAddress;
import java.time.Clock;
import java.time.Duration;
@@ -178,11 +179,22 @@ public class AssistantRuntime
this.contextBuilder = new ContextBuilder(config, kernel, toolbox).withKnowledgeTool(knowledgeTool).withBulkLoadTools(bulkLoadTools);
if(Boolean.TRUE.equals(config.getRegisterShutdownHook()))
{
///////////////////////////////////////////////////////////////////////
// the hook holds the runtime weakly: a runtime (and its QInstance) //
// that is dropped without shutdown() - every test that builds an //
// instance - must not stay reachable from the JVM's hooks //
///////////////////////////////////////////////////////////////////////
WeakReference<AssistantRuntime> self = new WeakReference<>(this);
Integer graceSeconds = config.getShutdownGraceSeconds();
shutdownHook = new Thread(() ->
{
try
{
shutdown(Duration.ofSeconds(config.getShutdownGraceSeconds()));
AssistantRuntime runtime = self.get();
if(runtime != null)
{
runtime.shutdown(Duration.ofSeconds(graceSeconds));
}
}
catch(Throwable ignored)
{
@@ -18,9 +18,11 @@ package com.kingsrook.qbits.assistant.engine;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import com.kingsrook.qbits.assistant.AssistantRuntime;
import com.kingsrook.qbits.assistant.AssistantSecurity;
import com.kingsrook.qbits.assistant.model.AssistantModel;
import com.kingsrook.qbits.assistant.model.AssistantProvider;
import com.kingsrook.qbits.assistant.provider.RegistryModelProvider;
@@ -34,6 +36,7 @@ import com.kingsrook.qbits.assistant.spi.ModelRequest;
import com.kingsrook.qbits.assistant.spi.ModelResponse;
import com.kingsrook.qbits.assistant.spi.StreamHandle;
import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
@@ -124,6 +127,16 @@ public final class ConnectionTest
** Probe health, then make and account for a minimal completion.
*******************************************************************************/
private static Result call(AssistantRuntime runtime, ProviderService.Active active, String userId)
{
return (call(runtime, active, userId, Instant.now().plusSeconds(runtime.getConfig().getProviderTestTimeoutSeconds()), true));
}
/*******************************************************************************
** Share the minimal call and TEST accounting with central model health.
*******************************************************************************/
private static Result call(AssistantRuntime runtime, ProviderService.Active active, String userId, Instant deadline, Boolean probe)
{
List<ModelMessage> messages = new ArrayList<>();
messages.add(ModelMessage.user(PING_PROMPT));
@@ -136,19 +149,27 @@ public final class ConnectionTest
UsageRecorder.Call usageCall = new UsageRecorder.Call(userId, null, null, active, UsageRecorder.PURPOSE_TEST);
ModelPricing pricing = runtime.getProviderService().pricing(active, active.model());
long started = System.nanoTime();
StreamHandle handle = null;
try
{
AssistantProviderType type = runtime.getConfig().getProviderTypes().get(active.providerType());
HealthResult health = type.checkHealth(runtime.getProviderService().settings(active.row()));
if(Boolean.FALSE.equals(health.healthy()))
if(probe)
{
throw (new ModelProviderException(ModelProviderException.Kind.NETWORK, null, health.message()));
AssistantProviderType type = runtime.getConfig().getProviderTypes().get(active.providerType());
HealthResult health = type.checkHealth(runtime.getProviderService().settings(active.row()));
if(Boolean.FALSE.equals(health.healthy()))
{
throw (new ModelProviderException(ModelProviderException.Kind.NETWORK, null, health.message()));
}
}
StreamHandle handle = active.provider().stream(request, text ->
if(!Instant.now().isBefore(deadline) || Thread.currentThread().isInterrupted())
{
return (new Result(STATUS_ERROR, null, null, "The test deadline passed."));
}
handle = active.provider().stream(request, text ->
{
// the reply text is not used
});
ModelResponse response = handle.await(Duration.ofSeconds(runtime.getConfig().getProviderTestTimeoutSeconds()));
ModelResponse response = handle.await(Duration.between(Instant.now(), deadline));
long latency = (System.nanoTime() - started) / 1_000_000;
UsageRecorder.save(runtime.getStore(), UsageRecorder.row(usageCall, response, null, latency, pricing));
if(ModelResponse.STOP_CANCELLED.equals(response.getStopReason()))
@@ -170,8 +191,45 @@ public final class ConnectionTest
catch(InterruptedException e)
{
Thread.currentThread().interrupt();
UsageRecorder.save(runtime.getStore(), UsageRecorder.row(usageCall, null,
new ModelProviderException(ModelProviderException.Kind.TIMEOUT, null, "The test was interrupted."),
(System.nanoTime() - started) / 1_000_000, pricing));
return (new Result(STATUS_ERROR, null, null, "The test was interrupted."));
}
finally
{
if(handle != null)
{
handle.cancel();
}
}
}
/*******************************************************************************
** Central health uses the same probe without changing manual lastTest fields.
** Purpose is a string, so health calls reuse TEST rather than adding an enum.
*******************************************************************************/
public static Result checkModel(AssistantRuntime runtime, AssistantModel model, Instant deadline) throws QException
{
if(!Instant.now().isBefore(deadline) || Thread.currentThread().isInterrupted())
{
return (new Result(STATUS_ERROR, null, null, "The test deadline passed."));
}
if(!model.isChat())
{
return (testEmbedding(runtime, model));
}
AssistantProvider row = runtime.getStore().getProvider(model.getProviderId());
var type = runtime.getConfig().getProviderTypes().get(row.getProviderType());
var settings = runtime.getProviderService().settings(row);
int seconds = (int) Math.max(1, Math.min(30, Duration.between(Instant.now(), deadline).toSeconds()));
settings.withConnectTimeoutSeconds(Math.min(settings.getConnectTimeoutSeconds(), seconds))
.withRequestTimeoutSeconds(Math.min(settings.getRequestTimeoutSeconds(), seconds));
var provider = new RegistryModelProvider(type.create(settings), model);
return (call(runtime, new ProviderService.Active(row, row.getProviderType(), model.getModelId(), provider),
AssistantSecurity.userId(QContext.getQSession()), deadline, false));
}
@@ -246,7 +304,7 @@ public final class ConnectionTest
try
{
new EmbeddingService(runtime).embed(model.getId(), List.of("Connection test."),
DataClassification.PUBLIC, "", null, null);
DataClassification.PUBLIC, "", null, null, UsageRecorder.PURPOSE_TEST);
return (new Result(STATUS_OK, (int) ((System.nanoTime() - started) / 1_000_000), model.getModelId(), null));
}
catch(QException e)
@@ -27,6 +27,8 @@ import com.kingsrook.qbits.assistant.spi.HealthResult;
import com.kingsrook.qbits.assistant.spi.ModelProviderException;
import com.kingsrook.qbits.assistant.spi.ProviderSettings;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.health.HealthRunner;
import com.kingsrook.qqq.backend.core.health.HealthStatus;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
@@ -72,6 +74,21 @@ public class ProviderHealth
{
return;
}
HealthRunner runner = HealthRunner.getForInstance(runtime.getQInstance());
String checkName = "assistant.provider." + row.getId();
Instant checked = runner == null ? null : runner.getLastOutcomeAt(checkName);
var outcome = runner == null ? null : runner.getLastOutcome(checkName);
if(checked != null && !checked.isAfter(now) && checked.plusSeconds(60).isAfter(now) && outcome != null)
{
if(outcome.getStatus() == HealthStatus.DOWN)
{
throw (new ModelProviderException(ModelProviderException.Kind.NETWORK, null, "Provider health check failed."));
}
if(outcome.getStatus() == HealthStatus.UP || outcome.getStatus() == HealthStatus.DEGRADED)
{
return;
}
}
ProviderSettings settings = runtime.getProviderService().settings(row).withConnectTimeoutSeconds(2).withRequestTimeoutSeconds(2);
FutureTask<HealthResult> probe = new FutureTask<>(() -> runtime.getConfig().getProviderTypes().get(row.getProviderType()).checkHealth(settings));
Thread.ofVirtual().name("assistant-health").start(probe);
@@ -481,13 +481,23 @@ public class ProviderService
** micro-dollars, from the usage rows the session can see.
*******************************************************************************/
public long monthToDateCostMicros(Integer providerId) throws QException
{
return (monthToDateCostMicros(providerId, false));
}
/*******************************************************************************
** Shared accounting permits health-runner caps without granting owner access.
*******************************************************************************/
public long monthToDateCostMicros(Integer providerId, Boolean sharedAccounting) throws QException
{
if(providerId == null)
{
return (0);
}
Instant monthStart = LocalDate.now(clock.withZone(ZoneOffset.UTC)).withDayOfMonth(1).atStartOfDay().toInstant(ZoneOffset.UTC);
return (store.sumUsageCostSince(providerId, monthStart));
return (store.sumUsageCostSince(providerId, monthStart, sharedAccounting));
}
@@ -0,0 +1,279 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.assistant.health;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import com.kingsrook.qbits.assistant.AssistantRuntime;
import com.kingsrook.qbits.assistant.AssistantSecurity;
import com.kingsrook.qbits.assistant.engine.ConnectionTest;
import com.kingsrook.qbits.assistant.engine.ProviderService;
import com.kingsrook.qbits.assistant.model.AssistantModel;
import com.kingsrook.qbits.assistant.model.AssistantProvider;
import com.kingsrook.qbits.assistant.spi.HealthResult;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.health.HealthCategory;
import com.kingsrook.qqq.backend.core.health.HealthCheck;
import com.kingsrook.qqq.backend.core.health.HealthCheckInput;
import com.kingsrook.qqq.backend.core.health.HealthCheckKind;
import com.kingsrook.qqq.backend.core.health.HealthCheckMetaData;
import com.kingsrook.qqq.backend.core.health.HealthCheckOutcome;
import com.kingsrook.qqq.backend.core.health.HealthCheckSupplier;
import com.kingsrook.qqq.backend.core.health.HealthCheckSupplierMetaData;
import com.kingsrook.qqq.backend.core.health.HealthChecksMetaData;
import com.kingsrook.qqq.backend.core.health.HealthDetails;
import com.kingsrook.qqq.backend.core.health.HealthStatus;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
/*******************************************************************************
** Central health adapters reuse the existing probes without exposing their text.
*******************************************************************************/
public final class AssistantHealthChecks
{
/*******************************************************************************
** Utility class.
*******************************************************************************/
private AssistantHealthChecks()
{
}
/*******************************************************************************
** Register dynamic namespaces once, even if production is repeated.
*******************************************************************************/
public static void register(QInstance instance)
{
HealthChecksMetaData registry = HealthChecksMetaData.ofOrWithNew(instance);
register(registry, "assistant.provider", Providers.class);
register(registry, "assistant.model", Models.class);
}
/*******************************************************************************
** Suppliers carry the component; individual definitions also declare it.
*******************************************************************************/
private static void register(HealthChecksMetaData registry, String prefix, Class<? extends HealthCheckSupplier> supplier)
{
if(registry.getSuppliers().stream().noneMatch(existing -> prefix.equals(existing.getNamePrefix())))
{
registry.withSupplier(new HealthCheckSupplierMetaData().withNamePrefix(prefix).withComponent("assistant").withSupplier(new QCodeReference(supplier)));
}
}
/*******************************************************************************
** Provider and model ids are integer primary keys, hence already lowercase.
*******************************************************************************/
private static HealthCheckMetaData definition(String name, Integer interval, Integer timeout, HealthCheck check)
{
return (new HealthCheckMetaData().withName(name).withLabel(name).withComponent("assistant")
.withKind(HealthCheckKind.DEPENDENCY).withCategory(HealthCategory.AI)
.withIntervalSeconds(interval).withTimeoutMs(timeout).withCheckInstance(check));
}
/*******************************************************************************
** One definition per enabled provider; no I/O to providers during resolution.
*******************************************************************************/
public static class Providers implements HealthCheckSupplier
{
/***************************************************************************
** Re-resolve rows when the central registry refreshes its suppliers.
***************************************************************************/
@Override
public List<HealthCheckMetaData> supply(QInstance instance) throws QException
{
AssistantRuntime runtime = AssistantRuntime.of(instance);
List<HealthCheckMetaData> checks = new ArrayList<>();
for(AssistantProvider row : runtime.getStore().listProviders())
{
if(Boolean.TRUE.equals(row.getEnabled()))
{
checks.add(definition("assistant.provider." + row.getId(), "LOCAL".equals(runtime.getProviderService().location(row)) ? 60 : 300, 10000,
input -> bounded(input, () -> provider(runtime, row.getId(), input))));
}
}
return (checks);
}
}
/*******************************************************************************
** One definition per enabled model, including models of disabled providers.
*******************************************************************************/
public static class Models implements HealthCheckSupplier
{
/***************************************************************************
** Eligibility is checked again when a scheduled check actually executes.
***************************************************************************/
@Override
public List<HealthCheckMetaData> supply(QInstance instance) throws QException
{
AssistantRuntime runtime = AssistantRuntime.of(instance);
List<HealthCheckMetaData> checks = new ArrayList<>();
for(AssistantModel row : runtime.getStore().listModels())
{
if(Boolean.TRUE.equals(row.getEnabled()))
{
checks.add(definition("assistant.model." + row.getId(), 21600, 30000,
input -> bounded(input, () -> model(runtime, row.getId(), input))));
}
}
return (checks);
}
}
/*******************************************************************************
** Clamp transport timeouts and bound even an adapter that ignores them.
** Keep the caller's session; never create or elevate a health-check session.
*******************************************************************************/
private static HealthCheckOutcome bounded(HealthCheckInput input, Callable<HealthCheckOutcome> call)
{
long remaining = Duration.between(Instant.now(), input.getDeadline()).toMillis();
if(remaining <= 0)
{
return (HealthCheckOutcome.down("Health check timed out"));
}
var session = QContext.getQSession();
FutureTask<HealthCheckOutcome> task = new FutureTask<>(() ->
{
QContext.init(input.getQInstance(), session);
try
{
return (call.call());
}
finally
{
QContext.clear();
}
});
Thread.ofVirtual().name("assistant-central-health").start(task);
try
{
return (task.get(remaining, TimeUnit.MILLISECONDS));
}
catch(InterruptedException e)
{
Thread.currentThread().interrupt();
return (HealthCheckOutcome.down("Health check interrupted"));
}
catch(java.util.concurrent.TimeoutException e)
{
return (HealthCheckOutcome.down("Health check timed out"));
}
catch(java.util.concurrent.ExecutionException e)
{
return (HealthCheckOutcome.down("Health check failed"));
}
finally
{
task.cancel(true);
}
}
/*******************************************************************************
** Unsupported probes may read recent traffic-produced breaker state only.
*******************************************************************************/
private static HealthCheckOutcome provider(AssistantRuntime runtime, Integer id, HealthCheckInput input) throws QException
{
AssistantProvider row = runtime.getStore().getProvider(id);
if(row == null || !Boolean.TRUE.equals(row.getEnabled()))
{
return (HealthCheckOutcome.unknown("Provider disabled or unavailable"));
}
var type = runtime.getConfig().getProviderTypes().get(row.getProviderType());
if(type == null)
{
return (HealthCheckOutcome.unknown("Provider type unavailable"));
}
var settings = runtime.getProviderService().settings(row);
int seconds = (int) Math.max(1, Math.min(10, Duration.between(Instant.now(), input.getDeadline()).toSeconds()));
settings.withConnectTimeoutSeconds(Math.min(settings.getConnectTimeoutSeconds(), seconds))
.withRequestTimeoutSeconds(Math.min(settings.getRequestTimeoutSeconds(), seconds));
HealthResult result = type.checkHealth(settings);
HealthCheckOutcome outcome;
if(result.healthy() == null)
{
Instant checked = row.getHealthCheckedAt();
Instant now = runtime.getClock().instant();
if(checked != null && !checked.isAfter(now) && checked.plusSeconds(600).isAfter(now) && List.of("UP", "DOWN", "DEGRADED").contains(row.getHealthState() == null ? "" : row.getHealthState()))
{
outcome = new HealthCheckOutcome().withStatus(HealthStatus.valueOf(row.getHealthState())).withMessage("Recent circuit-breaker state");
}
else
{
outcome = HealthCheckOutcome.unknown("health probe not supported");
}
}
else
{
outcome = result.healthy() ? HealthCheckOutcome.up("Provider reachable") : HealthCheckOutcome.down("Provider health probe failed");
}
return (outcome.withLatencyMs(result.latencyMs()).withDetails(new HealthDetails().withValue("providerId", id)));
}
/*******************************************************************************
** Reuse the minimal connection call and its accounting, without lastTest writes.
*******************************************************************************/
private static HealthCheckOutcome model(AssistantRuntime runtime, Integer id, HealthCheckInput input) throws QException
{
AssistantModel row = runtime.getStore().listModels().stream().filter(model -> id.equals(model.getId())).findFirst().orElse(null);
if(row == null || !Boolean.TRUE.equals(row.getEnabled()))
{
return (HealthCheckOutcome.unknown("Model disabled or unavailable"));
}
AssistantProvider provider = runtime.getStore().getProvider(row.getProviderId());
if(provider == null || !Boolean.TRUE.equals(provider.getEnabled()))
{
return (HealthCheckOutcome.unknown("Provider disabled or unavailable"));
}
Long cap = ProviderService.capMicros(provider);
if(cap != null && runtime.getProviderService().monthToDateCostMicros(provider.getId(), true) >= cap)
{
return (HealthCheckOutcome.unknown("skipped: cost cap"));
}
if(runtime.getPolicyService().current(true).killSwitch())
{
return (HealthCheckOutcome.unknown("skipped: kill switch"));
}
AssistantSecurity.ownerKey(QContext.getQSession(), runtime.getConfig());
ConnectionTest.Result result = ConnectionTest.checkModel(runtime, row, input.getDeadline());
return ((result.ok() ? HealthCheckOutcome.up("Model connection succeeded") : HealthCheckOutcome.down("Model connection failed"))
.withDetails(new HealthDetails().withValue("modelId", id).withValue("providerId", provider.getId()))
.withLatencyMs(result.latencyMs() == null ? null : result.latencyMs().longValue()));
}
}
@@ -0,0 +1,79 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.assistant.health;
import java.util.List;
import java.util.Map;
import com.kingsrook.qbits.assistant.metadata.AssistantMetaDataBuilder;
import com.kingsrook.qbits.assistant.tools.FilterLinkBuilder;
import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.AbstractWidgetRenderer;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.health.HealthChecksMetaData;
import com.kingsrook.qqq.backend.core.health.HealthRunner;
import com.kingsrook.qqq.backend.core.health.HealthStatus;
import com.kingsrook.qqq.backend.core.model.actions.widgets.RenderWidgetInput;
import com.kingsrook.qqq.backend.core.model.actions.widgets.RenderWidgetOutput;
import com.kingsrook.qqq.backend.core.model.dashboard.widgets.StatisticsData;
import com.kingsrook.qqq.backend.core.model.dashboard.widgets.WidgetType;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
import com.kingsrook.qqq.backend.core.model.metadata.dashboard.QWidgetMetaData;
/*******************************************************************************
** A single generic status card links to the central checks, without probing.
*******************************************************************************/
public class AssistantHealthWidget extends AbstractWidgetRenderer
{
public static final String NAME = "assistantHealth";
/*******************************************************************************
** No HTML, history or per-provider check list on the Assistant home.
*******************************************************************************/
public static QWidgetMetaData widget()
{
return (new QWidgetMetaData().withName(NAME).withLabel("Assistant health")
.withType(WidgetType.STATISTICS.getType()).withCodeReference(new QCodeReference(AssistantHealthWidget.class))
.withGridColumns(12).withPermissionRules(AssistantMetaDataBuilder.adminRules()));
}
/*******************************************************************************
** Summarize the runner's latest results; full history lives in System Health.
*******************************************************************************/
@Override
public RenderWidgetOutput render(RenderWidgetInput input) throws QException
{
var instance = QContext.getQInstance();
HealthRunner runner = HealthRunner.getForInstance(instance);
HealthStatus status = HealthStatus.UNKNOWN;
if(runner != null)
{
status = HealthStatus.aggregate(HealthChecksMetaData.ofOrWithNew(instance).resolveAll(instance).stream()
.filter(check -> "assistant".equals(check.getComponent()))
.map(check -> runner.getLastOutcome(check.getName()))
.map(outcome -> outcome == null ? HealthStatus.UNKNOWN : outcome.getStatus()).toArray(HealthStatus[]::new));
}
return (new RenderWidgetOutput(new StatisticsData().withCount(status.name())
.withCountContext("System Health")
.withCountURL(FilterLinkBuilder.href("health_check", List.of(Map.of("fieldName", "component", "operator", "EQUALS", "values", List.of("assistant")))))));
}
}
@@ -32,7 +32,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.dashboard.QWidgetMetaData;
/*******************************************************************************
** The Assistant app home: one full-width HTML widget, the single owner of the
** assistant's health, usage and cost data (no other dashboard repeats it).
** assistant's configuration, usage and cost data. Health lives in System Health.
*******************************************************************************/
public final class AssistantHome
{
@@ -17,16 +17,6 @@
package com.kingsrook.qbits.assistant.home;
import com.kingsrook.qbits.assistant.engine.BudgetService;
import com.kingsrook.qbits.assistant.metadata.AssistantRecordScreens;
import com.kingsrook.qbits.assistant.model.AssistantBudget;
import com.kingsrook.qbits.assistant.model.AssistantBudgetOverride;
import com.kingsrook.qbits.assistant.model.AssistantConversation;
import com.kingsrook.qbits.assistant.model.AssistantModelPrice;
import com.kingsrook.qbits.assistant.model.AssistantProvider;
import com.kingsrook.qbits.assistant.model.AssistantToolCall;
import com.kingsrook.qbits.assistant.model.AssistantUsage;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.text.NumberFormat;
@@ -38,6 +28,16 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import com.kingsrook.qbits.assistant.engine.BudgetService;
import com.kingsrook.qbits.assistant.metadata.AssistantRecordScreens;
import com.kingsrook.qbits.assistant.model.AssistantBudget;
import com.kingsrook.qbits.assistant.model.AssistantBudgetOverride;
import com.kingsrook.qbits.assistant.model.AssistantConversation;
import com.kingsrook.qbits.assistant.model.AssistantModelPrice;
import com.kingsrook.qbits.assistant.model.AssistantProvider;
import com.kingsrook.qbits.assistant.model.AssistantToolCall;
import com.kingsrook.qbits.assistant.model.AssistantUsage;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
/*******************************************************************************
@@ -159,44 +159,17 @@ public final class AssistantHomeView
/*******************************************************************************
** The status row: is the assistant answering, with what, at what cost.
** Configuration and usage; health status belongs to the central card.
*******************************************************************************/
private static void status(StringBuilder html, AssistantHomeData data, String prefix)
{
open(html, "Health");
open(html, "Configuration and usage");
AssistantHomeData.ProviderHealth active = data.getProviders().stream().filter(AssistantHomeData.ProviderHealth::active).findFirst().orElse(null);
html.append("<div style=\"display:flex;flex-wrap:wrap;gap:.5rem .75rem;align-items:center;\">");
if(data.isKillSwitch())
{
html.append(chip(RED, "Turned off by the kill switch"));
}
else if(data.getProviders().isEmpty())
{
html.append(chip(GREY, "No provider configured here (the application's built-in provider is used, if any)"));
}
else if(active == null)
{
html.append(chip(RED, "No provider is turned on"));
}
else if(!active.keyPresent() || !active.problems().isEmpty())
{
html.append(chip(RED, "Default provider cannot be used"));
}
else if("ERROR".equals(active.row().getLastTestStatus()))
{
html.append(chip(AMBER, "Last connection test failed"));
}
else
{
html.append(chip(GREEN, "Answering"));
}
html.append("</div><div style=\"").append(CELLS).append("\">");
html.append("<div style=\"").append(CELLS).append("\">");
cell(html, "Provider", active == null ? "None" : active.row().getName() + " (" + nvl(active.row().getModel(), "default model") + ")");
cell(html, "HIPAA mode", Boolean.TRUE.equals(data.getHipaaMode()) ? "On" : "Off");
cell(html, "PHI retention", data.getHipaaRetentionDays() == null ? "Normal retention" : data.getHipaaRetentionDays() + " days");
cell(html, "API key", active == null ? "Not applicable" : active.keyStatus());
cell(html, "Last test", active == null ? "Not applicable" : lastTest(active.row()));
Integer rate = data.errorRatePercent7d();
cell(html, "Error rate, 7 days", rate == null ? "No requests yet" : rate + "% of " + number(data.getCalls7d()) + " requests");
cell(html, "Cost this month", money(data.getMonthCostMicros()) + (data.getInstanceMonthlyCapMicros() == null ? "" : " of " + money(data.getInstanceMonthlyCapMicros()) + " limit"));
@@ -234,17 +207,8 @@ public final class AssistantHomeView
{
html.append(chip(GREY, "Off"));
}
if(row.getLastTestStatus() != null)
{
html.append(chip("OK".equals(row.getLastTestStatus()) ? GREEN : RED, "OK".equals(row.getLastTestStatus()) ? "Last test passed" : "Last test failed"));
}
html.append("</div><div style=\"").append(CELLS).append("\">");
cell(html, "Type", nvl(p.typeLabel(), "Not set"));
cell(html, "Health", nvl(row.getHealthState(), "Unknown"));
if(row.getDownUntil() != null)
{
cell(html, "Down until", when(row.getDownUntil()));
}
cell(html, "Kind", AssistantRecordScreens.providerKindLabel(p.kind()));
cell(html, "Location", "LOCAL".equals(p.location()) ? "Local" : "Cloud");
if(Boolean.TRUE.equals(row.getHipaaEligible()))
@@ -254,7 +218,6 @@ public final class AssistantHomeView
cell(html, "Enabled models", String.valueOf(p.enabledModels()));
cell(html, "Model", nvl(row.getModel(), "Default model"));
cell(html, "API key", p.keyStatus());
cell(html, "Last test", lastTest(row));
cell(html, "Cost this month", money(p.monthCostMicros()) + (p.capMicros() == null ? ", no cap" : " of " + money(p.capMicros()) + " cap"));
List<String> limits = new ArrayList<>();
if(row.getMaxOutputTokensPerRequest() != null)
@@ -502,24 +465,6 @@ public final class AssistantHomeView
/*******************************************************************************
** A provider's last test in words.
*******************************************************************************/
static String lastTest(AssistantProvider row)
{
if(row.getLastTestAt() == null)
{
return ("Not tested yet: run Test connection");
}
if("OK".equals(row.getLastTestStatus()))
{
return ("Passed " + WHEN.format(row.getLastTestAt()) + ", " + nvl(row.getLastTestModel(), "model") + " in " + row.getLastTestLatencyMs() + " ms");
}
return ("Failed " + WHEN.format(row.getLastTestAt()) + ": " + nvl(row.getLastTestError(), "no detail"));
}
/*******************************************************************************
** Micro-dollars as dollars ("$0.00", "<$0.01" for a little, "$12.34").
*******************************************************************************/
@@ -17,9 +17,14 @@
package com.kingsrook.qbits.assistant.jobs;
import com.kingsrook.qqq.backend.core.exceptions.QException;
/*******************************************************************************
** STAND-IN(core=QRun-IO/qqq#1210, removeBy=RC.9): core job tracking
** (jobs_job tracking rows) will implement this contract; hosts may wire their own.
** Publishes assistant answers as jobs. The default, JobTrackingRunTracker,
** uses QQQ core job tracking (the Jobs tray); NOOP turns it off, and hosts
** may wire their own. Every call is isolated from the turn: a failure here
** never affects the answer.
*******************************************************************************/
public interface AssistantRunTracker
{
@@ -54,7 +59,7 @@ public interface AssistantRunTracker
/*******************************************************************************
** Publish a run and return its tracking id.
*******************************************************************************/
String start(String userId, Integer turnId, Integer conversationId, String label);
String start(String userId, Integer turnId, Integer conversationId, String label) throws QException;
/*******************************************************************************
** Publish a changed phase.
@@ -64,12 +69,12 @@ public interface AssistantRunTracker
/*******************************************************************************
** Publish the terminal outcome.
*******************************************************************************/
void finish(String trackingId, String status, String message);
void finish(String trackingId, String status, String message) throws QException;
/*******************************************************************************
** Wire the host's Stop action to the owner's cancellation path.
*******************************************************************************/
default void onCancel(String trackingId, Runnable cancel)
default void onCancel(String trackingId, Runnable cancel) throws QException
{
}
}
@@ -0,0 +1,170 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.assistant.jobs;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import com.kingsrook.qbits.assistant.model.AssistantConversation;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.model.JobStatus;
import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
import com.kingsrook.qqq.backend.core.jobs.progress.JobTracking;
import com.kingsrook.qqq.backend.core.jobs.progress.TrackedJobRequest;
import com.kingsrook.qqq.backend.core.jobs.runtime.DurableJobsRuntime;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
/*******************************************************************************
** Publishes each assistant answer to the Jobs tray as a QQQ tracking job
** (core JobTracking, qqq#1210): processName "assistant.turn", phases
** thinking / tool / writing, the tool's label as the message, and the tray's
** Stop wired to the turn's own Stop. Never prompt or answer text. When the
** durable jobs runtime is not running on this pod, answers are simply not
** tracked.
*******************************************************************************/
public class JobTrackingRunTracker implements AssistantRunTracker
{
public static final String PROCESS_NAME = "assistant.turn";
public static final String LABEL = "Assistant answer";
private final Map<String, JobProgress> reporters = new ConcurrentHashMap<>();
/*******************************************************************************
** Start a tracking job linked to the conversation, or null when the jobs
** runtime is not running here.
*******************************************************************************/
@Override
public String start(String userId, Integer turnId, Integer conversationId, String label) throws QException
{
QInstance qInstance = QContext.getQInstance();
if(qInstance == null || DurableJobsRuntime.get(qInstance) == null)
{
return (null);
}
Integer jobId = JobTracking.start(new TrackedJobRequest()
.withUserId(userId)
.withLabel(LABEL)
.withProcessName(PROCESS_NAME)
.withTableName(AssistantConversation.TABLE_NAME)
.withRecordId(conversationId));
String trackingId = String.valueOf(jobId);
reporters.put(trackingId, JobTracking.progress(jobId).phases(
JobPhase.of("thinking", "Thinking", 1),
JobPhase.of("tool", "Using a tool", 1),
JobPhase.of("writing", "Writing", 1)));
return (trackingId);
}
/*******************************************************************************
** The current phase, with the tool's label as the message.
*******************************************************************************/
@Override
public void progress(String trackingId, String phase, String message)
{
JobProgress reporter = trackingId == null ? null : reporters.get(trackingId);
if(reporter == null || !("thinking".equals(phase) || "tool".equals(phase) || "writing".equals(phase)))
{
return;
}
reporter.phase(phase);
if("tool".equals(phase) && message != null)
{
reporter.message(message);
reporter.flush();
}
}
/*******************************************************************************
** COMPLETE -> COMPLETED, CANCELLED -> CANCELLED, anything else -> FAILED.
*******************************************************************************/
@Override
public void finish(String trackingId, String status, String message) throws QException
{
if(trackingId == null)
{
return;
}
JobProgress reporter = reporters.remove(trackingId);
if(reporter != null)
{
reporter.flush();
}
JobTracking.finish(Integer.valueOf(trackingId), jobStatus(status), finishMessage(status));
}
/*******************************************************************************
** The tray's Stop runs on this pod and stops the turn; the turn's own
** finaliser then finishes the job as CANCELLED.
*******************************************************************************/
@Override
public void onCancel(String trackingId, Runnable cancel) throws QException
{
if(trackingId != null)
{
JobTracking.onCancel(Integer.valueOf(trackingId), cancel);
}
}
/*******************************************************************************
** The jobs status for an assistant turn status.
*******************************************************************************/
static JobStatus jobStatus(String status)
{
if("COMPLETE".equals(status))
{
return (JobStatus.COMPLETED);
}
if("CANCELLED".equals(status))
{
return (JobStatus.CANCELLED);
}
return (JobStatus.FAILED);
}
/*******************************************************************************
** A fixed message (never the answer, the prompt or a provider error).
*******************************************************************************/
static String finishMessage(String status)
{
if("COMPLETE".equals(status))
{
return ("Answered.");
}
if("CANCELLED".equals(status))
{
return ("Stopped.");
}
return ("The answer did not finish.");
}
}
@@ -1,126 +0,0 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.assistant.metadata;
import java.io.IOException;
import java.io.InputStream;
import java.util.LinkedHashMap;
import java.util.Map;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.help.QHelpContent;
import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
/*******************************************************************************
** Applies help/qbit-assistant.yaml (the QQQ help-bundle format) to the
** Assistant app and its assistant_* tables: what each table is for and how
** its rows get created, shown on the table screen and when it is empty.
** STAND-IN(core=#1198): replace with HelpBundles.register once the pinned
** QQQ release includes it; the bundle file stays the same.
*******************************************************************************/
public final class AssistantHelp
{
public static final String RESOURCE = "/help/qbit-assistant.yaml";
public static final String SLOT = "help";
/*******************************************************************************
** Utility class.
*******************************************************************************/
private AssistantHelp()
{
}
/*******************************************************************************
** The bundle's entries by key, validated (fails startup if malformed).
*******************************************************************************/
public static Map<String, JsonNode> entries() throws QException
{
try(InputStream stream = AssistantHelp.class.getResourceAsStream(RESOURCE))
{
if(stream == null)
{
throw (new IOException("missing resource"));
}
JsonNode bundle = new ObjectMapper(new YAMLFactory()).readTree(stream);
if(!"qbit-assistant".equals(bundle.path("bundle").asText()) || !"QBIT".equals(bundle.path("layer").asText()))
{
throw (new IOException("invalid bundle header"));
}
Map<String, JsonNode> entries = new LinkedHashMap<>();
for(JsonNode entry : bundle.path("entries"))
{
String key = entry.path("key").asText();
String summary = entry.path("summary").asText();
if(key.isBlank() || summary.isBlank() || summary.length() > 120 || entry.path("content").asText().isBlank() || entries.put(key, entry) != null)
{
throw (new IOException("invalid or duplicate entry: " + key));
}
}
return (entries);
}
catch(IOException | RuntimeException e)
{
throw (new QException("Could not load " + RESOURCE, e));
}
}
/*******************************************************************************
** Apply the help to the app and every assistant_* table in the instance.
** A host's own help on a table is kept (the bundle only fills the slot).
*******************************************************************************/
public static void apply(QInstance qInstance) throws QException
{
Map<String, JsonNode> entries = entries();
for(QTableMetaData table : qInstance.getTables().values())
{
JsonNode entry = entries.get("table:" + table.getName() + ";slot:" + SLOT);
if(entry != null && (table.getHelpContent() == null || !table.getHelpContent().containsKey(SLOT)))
{
table.withHelpContent(SLOT, content(entry));
}
}
QAppMetaData app = qInstance.getApp(AssistantMetaDataBuilder.APP_NAME);
JsonNode appEntry = entries.get("app:" + AssistantMetaDataBuilder.APP_NAME + ";slot:" + SLOT);
if(app != null && appEntry != null && (app.getHelpContent() == null || !app.getHelpContent().containsKey(SLOT)))
{
app.withHelpContent(SLOT, content(appEntry));
}
}
/*******************************************************************************
** Summary as the first paragraph, then the content, as markdown.
*******************************************************************************/
static QHelpContent content(JsonNode entry)
{
return (new QHelpContent().withContentAsMarkdown(entry.path("summary").asText().strip() + "\n\n" + entry.path("content").asText().strip()));
}
}
@@ -20,6 +20,7 @@ package com.kingsrook.qbits.assistant.metadata;
import java.util.ArrayList;
import java.util.List;
import com.kingsrook.qbits.assistant.AssistantQBitConfig;
import com.kingsrook.qbits.assistant.health.AssistantHealthWidget;
import com.kingsrook.qbits.assistant.home.AssistantAccessWidget;
import com.kingsrook.qbits.assistant.home.AssistantHome;
import com.kingsrook.qbits.assistant.model.AssistantBudget;
@@ -383,7 +384,7 @@ public class AssistantMetaDataBuilder
.withLabel("Assistant")
.withIcon(new QIcon("smart_toy"))
.withPermissionRules(adminRules())
.withWidgets(List.of(AssistantHome.WIDGET_NAME, AssistantAccessWidget.WIDGET_NAME));
.withWidgets(List.of(AssistantHealthWidget.NAME, AssistantHome.WIDGET_NAME, AssistantAccessWidget.WIDGET_NAME));
for(String name : NAV_TABLES)
{
tables.stream().filter(t -> t.getName().equals(name)).findFirst().ifPresent(app::withChild);
@@ -814,11 +814,22 @@ public class AssistantStore
** The cost (micro-dollars) of a provider's calls since a time.
*******************************************************************************/
public long sumUsageCostSince(Integer providerId, Instant since) throws QException
{
return (sumUsageCostSince(providerId, since, false));
}
/*******************************************************************************
** Automated health uses the existing private accounting projection so a cap
** includes all users, without reading their owner-locked usage records.
*******************************************************************************/
public long sumUsageCostSince(Integer providerId, Instant since, Boolean sharedAccounting) throws QException
{
long total = 0;
for(QRecord record : queryRecords(AssistantUsage.TABLE_NAME, new QQueryFilter()
for(QRecord record : queryRecords(sharedAccounting ? AssistantBudgetUsage.TABLE_NAME : AssistantUsage.TABLE_NAME, new QQueryFilter()
.withCriteria("providerId", QCriteriaOperator.EQUALS, providerId)
.withCriteria("createDate", QCriteriaOperator.GREATER_THAN_OR_EQUALS, since), MAX_SUM_ROWS))
.withCriteria(sharedAccounting ? "occurredAt" : "createDate", QCriteriaOperator.GREATER_THAN_OR_EQUALS, since), MAX_SUM_ROWS))
{
Long cost = record.getValueLong("costMicros");
total += cost == null ? 0 : cost;
@@ -64,11 +64,21 @@ public class EmbeddingService
** including failures, and updates the existing daily budget rollups.
*******************************************************************************/
public List<float[]> embed(Integer modelId, List<String> texts, DataClassification classification, String tables, Integer conversationId, Integer turnId) throws QException
{
return (embed(modelId, texts, classification, tables, conversationId, turnId, UsageRecorder.PURPOSE_EMBED));
}
/*******************************************************************************
** Connection diagnostics use TEST accounting with the same embedding guards.
*******************************************************************************/
public List<float[]> embed(Integer modelId, List<String> texts, DataClassification classification, String tables, Integer conversationId, Integer turnId, String purpose) throws QException
{
List<float[]> vectors = new ArrayList<>();
for(Integer offset = 0; offset < texts.size(); offset += 64)
{
vectors.addAll(batch(modelId, texts.subList(offset, Math.min(offset + 64, texts.size())), classification, tables, conversationId, turnId));
vectors.addAll(batch(modelId, texts.subList(offset, Math.min(offset + 64, texts.size())), classification, tables, conversationId, turnId, purpose));
}
return (vectors);
}
@@ -104,7 +114,7 @@ public class EmbeddingService
/*******************************************************************************
** Never downgrade an embedding call into a different vector space.
*******************************************************************************/
private List<float[]> batch(Integer modelId, List<String> texts, DataClassification classification, String tables, Integer conversationId, Integer turnId) throws QException
private List<float[]> batch(Integer modelId, List<String> texts, DataClassification classification, String tables, Integer conversationId, Integer turnId, String purpose) throws QException
{
String userId = AssistantSecurity.userId(QContext.getQSession());
ScheduledTurn.checkCost(runtime, conversationId, turnId);
@@ -140,6 +150,10 @@ public class EmbeddingService
ModelProviderException error = null;
try
{
if(Thread.currentThread().isInterrupted())
{
throw (new ModelProviderException(ModelProviderException.Kind.TIMEOUT, null, "Embedding request interrupted."));
}
EmbeddingResponse embedded = active.provider().embed(new EmbeddingRequest(active.model(), texts));
response = new ModelResponse().withModel(active.model()).withRequestId(embedded.requestId()).withUsage(embedded.usage());
if(embedded.vectors().size() != texts.size())
@@ -160,7 +174,7 @@ public class EmbeddingService
gate.release();
}
var pricing = runtime.getProviderService().pricing(active, active.model());
runtime.getStore().insert(UsageRecorder.row(new UsageRecorder.Call(userId, conversationId, turnId, active, UsageRecorder.PURPOSE_EMBED),
runtime.getStore().insert(UsageRecorder.row(new UsageRecorder.Call(userId, conversationId, turnId, active, purpose),
response, error, (System.nanoTime() - started) / 1_000_000, pricing));
var usage = response == null ? (error == null ? null : error.getPartialUsage()) : response.getUsage();
if(usage != null)
File diff suppressed because it is too large. Load diff
@@ -91,7 +91,7 @@ class AssistantQBitProducerTest extends BaseAssistantTest
//////////////////////////////////////////////////////////////////////
assertThat(qInstance.getApp(AssistantMetaDataBuilder.APP_NAME).getChildren()).extracting(c -> c.getName())
.containsExactly("assistant_scheduled_task", "assistant_rag_source", "assistant_budget", "assistant_provider", "assistant_profile", "assistant_skill", "assistant_data_access", "assistant_conversation", "assistant_model_price", "assistant_policy");
assertThat(qInstance.getApp(AssistantMetaDataBuilder.APP_NAME).getWidgets()).containsExactly("assistantHome", "assistantAccess");
assertThat(qInstance.getApp(AssistantMetaDataBuilder.APP_NAME).getWidgets()).containsExactly("assistantHealth", "assistantHome", "assistantAccess");
assertThat(qInstance.getProcess(AssistantMetaDataBuilder.RETENTION_PROCESS_NAME).getTableName()).isEqualTo("assistant_conversation");
assertThat(qInstance.getProcess(AssistantMetaDataBuilder.RETENTION_PROCESS_NAME)).isNotNull();
assertThat(qInstance.getProcess(AssistantMetaDataBuilder.RETENTION_PROCESS_NAME).getSchedule()).isNull();
@@ -26,11 +26,13 @@ import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Stream;
import com.kingsrook.qbits.assistant.metadata.AssistantHelp;
import com.kingsrook.qbits.assistant.metadata.AssistantMetaDataBuilder;
import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
import com.kingsrook.qqq.backend.core.model.metadata.help.QHelpContent;
import com.kingsrook.qqq.backend.core.actions.help.HelpTargetCatalog;
import com.kingsrook.qqq.backend.core.model.metadata.help.bundles.HelpBundles;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
import com.kingsrook.qqq.backend.core.testing.HelpBundleAssertions;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -121,21 +123,45 @@ class AssistantTablesGuardrailTest extends BaseAssistantTest
/*******************************************************************************
** Every table and the app have bundle help that says how rows get created.
** Every owned target has valid bundle help, even on a host without help storage.
*******************************************************************************/
@Test
void everyTableHasEmptyStateHelp() throws Exception
void everyOwnedTargetHasBundleHelp()
{
assertThat(AssistantHelp.entries()).allSatisfy((key, entry) -> assertThat(entry.path("summary").asText().length()).as(key).isLessThanOrEqualTo(120));
assertThat(qInstance.getTable("helpContent")).isNull();
var bundles = HelpBundles.of(qInstance).getBundles();
assertThat(bundles).hasSize(1);
assertThat(bundles.get(0).getId()).isEqualTo("qbit-assistant");
HelpBundleAssertions.forInstance(qInstance)
.bundles("help/qbit-assistant.yaml")
// Core-generated bulk actions belong to core's bundle, not this QBit.
.owning(target -> target.owner() instanceof QProcessMetaData process
? qInstance.getTable("assistant_conversation").getSourceQBitName().equals(process.getSourceQBitName())
: target.name() != null && target.name().startsWith("assistant"))
.assertComplete();
// RC.7's gate requires tables, fields, apps and processes; also cover our selectors and widgets.
for(var target : HelpTargetCatalog.collect(qInstance))
{
if(List.of("widget", "possibleValueSource").contains(target.kind()) && target.name().startsWith("assistant"))
{
assertThat(bundles.get(0).getEntries()).as(target.key())
.anyMatch(entry -> entry.getKey().equals(target.key()) && entry.getSummary() != null && !entry.getSummary().isBlank());
}
}
for(String table : assistantTables())
{
List<QHelpContent> help = qInstance.getTable(table).getHelpContent() == null ? null : qInstance.getTable(table).getHelpContent().get(AssistantHelp.SLOT);
assertThat(help).as(table + " table help").isNotEmpty();
assertThat(help.get(0).getContent()).as(table + " help says how rows get created").contains("How rows get created");
assertThat(bundles.get(0).getEntries())
.anySatisfy(entry ->
{
assertThat(entry.getKey()).isEqualTo("table:" + table + ";slot:help");
assertThat(entry.getContent()).isNotBlank();
});
}
assertThat(qInstance.getApp(AssistantMetaDataBuilder.APP_NAME).getHelpContent()).containsKey(AssistantHelp.SLOT);
assertThat(AssistantHelp.entries().keySet()).allMatch(key -> !key.startsWith("table:") || qInstance.getTable(key.substring(6, key.indexOf(';'))) != null,
"help names only real tables");
assertThat(bundles.get(0).getEntries()).anySatisfy(entry ->
{
assertThat(entry.getKey()).isEqualTo("app:" + AssistantMetaDataBuilder.APP_NAME + ";slot:help");
assertThat(entry.getContent()).isNotBlank();
});
}
@@ -676,7 +676,7 @@ class RoutingFallbackTest extends BaseAssistantTest
assertThat(script("a").getRequests()).isEmpty();
assertThat(store().getProvider(a.getProviderId()).getHealthState()).isEqualTo("DOWN");
String html = AssistantHomeView.render(AssistantHomeData.load(config.getRuntime()), "");
assertThat(html).contains("DOWN", "Down until", "0 fallbacks");
assertThat(html).contains("0 fallbacks").doesNotContain("Down until", "Last test failed");
}
@@ -0,0 +1,617 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.assistant.health;
import java.math.BigDecimal;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.kingsrook.qbits.assistant.AssistantQBitConfig;
import com.kingsrook.qbits.assistant.engine.ProviderHealth;
import com.kingsrook.qbits.assistant.engine.ProviderService;
import com.kingsrook.qbits.assistant.engine.UsageRecorder;
import com.kingsrook.qbits.assistant.model.AssistantModel;
import com.kingsrook.qbits.assistant.model.AssistantPolicy;
import com.kingsrook.qbits.assistant.model.AssistantProvider;
import com.kingsrook.qbits.assistant.model.AssistantUsage;
import com.kingsrook.qbits.assistant.provider.fake.ScriptedModelProvider;
import com.kingsrook.qbits.assistant.spi.AssistantModelProvider;
import com.kingsrook.qbits.assistant.spi.AssistantProviderType;
import com.kingsrook.qbits.assistant.spi.EmbeddingRequest;
import com.kingsrook.qbits.assistant.spi.EmbeddingResponse;
import com.kingsrook.qbits.assistant.spi.HealthResult;
import com.kingsrook.qbits.assistant.spi.ModelProviderException;
import com.kingsrook.qbits.assistant.spi.ModelUsage;
import com.kingsrook.qbits.assistant.spi.ProviderSettings;
import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.health.HealthCategory;
import com.kingsrook.qqq.backend.core.health.HealthCheckInput;
import com.kingsrook.qqq.backend.core.health.HealthCheckKind;
import com.kingsrook.qqq.backend.core.health.HealthCheckMetaData;
import com.kingsrook.qqq.backend.core.health.HealthCheckOutcome;
import com.kingsrook.qqq.backend.core.health.HealthChecksMetaData;
import com.kingsrook.qqq.backend.core.health.HealthRunner;
import com.kingsrook.qqq.backend.core.health.HealthStatus;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.actions.widgets.RenderWidgetInput;
import com.kingsrook.qqq.backend.core.model.dashboard.widgets.StatisticsData;
import com.kingsrook.qqq.backend.core.model.dashboard.widgets.WidgetType;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.model.session.QUser;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
/*******************************************************************************
** Registration, deadlines, billing and secret safety through the memory backend.
*******************************************************************************/
class AssistantHealthChecksTest extends BaseAssistantTest
{
private static final String SECRET = "S3NT1NEL-secret";
private final FakeType type = new FakeType();
private Boolean healthy = true;
private Boolean failProbe = false;
private Boolean failCreate = false;
private Boolean failEmbedding = false;
private Integer probes = 0;
private Integer embeddings = 0;
private ProviderSettings lastSettings;
private CountDownLatch probeWait;
private final CountDownLatch interrupted = new CountDownLatch(1);
/*******************************************************************************
** Resolve a poisoned secret through the normal key-reference mechanism.
*******************************************************************************/
@Override
protected void customizeConfig(AssistantQBitConfig config)
{
provider = new EmbeddingProvider();
config.withModelProvider(provider).withProviderType(type).withAllowedApiKeyEnvNames(Set.of("HEALTH_TEST_KEY"))
.withSecretResolver(name -> SECRET);
}
/*******************************************************************************
** An adapter with no network and controllable provider failures.
*******************************************************************************/
class FakeType implements AssistantProviderType
{
/***************************************************************************
** Fake provider contract.
***************************************************************************/
@Override
public String getType()
{
return ("health-fake");
}
/***************************************************************************
** Fake provider contract.
***************************************************************************/
@Override
public String getLabel()
{
return ("Health fake");
}
/***************************************************************************
** Fake provider contract.
***************************************************************************/
@Override
public List<String> getModels()
{
return (List.of("model"));
}
/***************************************************************************
** Fake provider contract.
***************************************************************************/
@Override
public String getDefaultApiKeyEnvName()
{
return ("HEALTH_TEST_KEY");
}
/***************************************************************************
** Fake provider contract.
***************************************************************************/
@Override
public String getDefaultBaseUrl()
{
return ("https://health.invalid");
}
/***************************************************************************
** Fake provider contract.
***************************************************************************/
@Override
public String getDefaultLocation(String kind)
{
return ("LOCAL");
}
/***************************************************************************
** Reuse the existing scripted completion fake.
***************************************************************************/
@Override
public AssistantModelProvider create(String key, String url)
{
assertThat(key).isEqualTo(SECRET);
if(failCreate)
{
throw (new IllegalStateException(SECRET));
}
return (provider);
}
/***************************************************************************
** Verify bounded settings and exercise hostile exception messages.
***************************************************************************/
@Override
public HealthResult checkHealth(ProviderSettings settings)
{
probes++;
lastSettings = settings;
assertThat(settings.getApiKey()).isEqualTo(SECRET);
if(failProbe)
{
throw (new IllegalStateException(SECRET));
}
if(probeWait != null)
{
try
{
probeWait.await();
}
catch(InterruptedException e)
{
interrupted.countDown();
Thread.currentThread().interrupt();
}
}
return (new HealthResult(healthy, 7L, SECRET));
}
}
/*******************************************************************************
** Keep embedding tests on the existing scripted-provider fixture as well.
*******************************************************************************/
class EmbeddingProvider extends ScriptedModelProvider
{
/***************************************************************************
** Minimal embedding result; no external service is contacted.
***************************************************************************/
@Override
public EmbeddingResponse embed(EmbeddingRequest request) throws ModelProviderException
{
embeddings++;
if(failEmbedding)
{
throw (new ModelProviderException(ModelProviderException.Kind.NETWORK, null, SECRET));
}
return (new EmbeddingResponse(List.of(new float[] { 1, 0 }), new ModelUsage().withInputTokens(2L), "fake"));
}
}
/*******************************************************************************
** Set up real settings records in the existing admin fixture session.
*******************************************************************************/
private AssistantProvider row() throws Exception
{
QContext.init(qInstance, AssistantTestFixture.sessionFor(AssistantTestFixture.ADMIN));
return (config.getRuntime().getStore().insert(new AssistantProvider().withName("Health test")
.withProviderType(type.getType()).withEnabled(true)));
}
/*******************************************************************************
** Enabled chat model with a stable integer id.
*******************************************************************************/
private AssistantModel model(AssistantProvider row) throws Exception
{
return (config.getRuntime().getStore().insert(new AssistantModel().withProviderId(row.getId()).withModelId("model").withEnabled(true)));
}
/*******************************************************************************
** Exercise the producer-installed suppliers, not a separately constructed check.
*******************************************************************************/
private HealthCheckMetaData check(String name) throws Exception
{
return (HealthChecksMetaData.ofOrWithNew(qInstance).resolveAll(qInstance).stream()
.filter(check -> name.equals(check.getName())).findFirst().orElseThrow());
}
/*******************************************************************************
** Raw outcomes must be clean; core's sanitize method is package-private.
*******************************************************************************/
private HealthCheckOutcome run(HealthCheckMetaData check) throws Exception
{
HealthCheckOutcome outcome = check.getCheckInstance().run(new HealthCheckInput().withQInstance(qInstance).withDeadline(Instant.now().plusSeconds(5)));
assertThat(outcome.getMessage()).doesNotContain(SECRET, "health.invalid", "Connection test.", "\n");
assertThat(outcome.getDetails().toJson()).doesNotContain(SECRET, "health.invalid", "Connection test.");
return (outcome);
}
/*******************************************************************************
** Both suppliers are installed by production, refresh enabled sets and use AI.
*******************************************************************************/
@Test
void registration() throws Exception
{
AssistantProvider local = row();
AssistantProvider cloud = config.getRuntime().getStore().insert(new AssistantProvider().withName("Cloud")
.withProviderType(type.getType()).withEnabled(true).withLocation("CLOUD"));
AssistantModel model = model(local);
model(cloud);
config.getRuntime().getStore().insert(new AssistantProvider().withName("Disabled").withProviderType(type.getType()).withEnabled(false));
config.getRuntime().getStore().insert(new AssistantModel().withProviderId(local.getId()).withModelId("disabled").withEnabled(false));
var registry = HealthChecksMetaData.ofOrWithNew(qInstance);
AssistantHealthChecks.register(qInstance);
assertThat(registry.getSuppliers()).hasSize(2).allSatisfy(supplier -> assertThat(supplier.getComponent()).isEqualTo("assistant"));
assertThat(registry.resolveAll(qInstance)).hasSize(4).allSatisfy(check ->
{
assertThat(check.getKind()).isEqualTo(HealthCheckKind.DEPENDENCY);
assertThat(check.getCategory()).isEqualTo(HealthCategory.AI);
assertThat(check.getComponent()).isEqualTo("assistant");
assertThat(check.getName()).matches("[a-z0-9]+([._-][a-z0-9]+)*");
});
assertThat(check("assistant.provider." + local.getId()).getIntervalSeconds()).isEqualTo(60);
assertThat(check("assistant.provider." + cloud.getId()).getIntervalSeconds()).isEqualTo(300);
var modelCheck = check("assistant.model." + model.getId());
assertThat(modelCheck.getIntervalSeconds()).isEqualTo(21600);
assertThat(modelCheck.getTimeoutMs()).isEqualTo(30000);
assertThat(check("assistant.provider." + local.getId()).getTimeoutMs()).isEqualTo(10000);
config.getRuntime().getStore().updateFields(AssistantModel.TABLE_NAME, new QRecord().withValue("id", model.getId()).withValue("enabled", false));
assertThat(registry.resolveAll(qInstance)).hasSize(3);
assertThat(run(modelCheck).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
}
/*******************************************************************************
** HealthResult messages and thrown adapter messages never enter outcomes.
*******************************************************************************/
@Test
void providerStatesAndNoSecrets() throws Exception
{
AssistantProvider row = row();
var check = check("assistant.provider." + row.getId());
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.UP);
assertThat(lastSettings.getRequestTimeoutSeconds()).isLessThanOrEqualTo(5);
healthy = false;
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.DOWN);
failProbe = true;
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.DOWN);
failProbe = false;
healthy = null;
assertThat(run(check).getMessage()).isEqualTo("health probe not supported");
for(String state : List.of("UP", "DOWN", "DEGRADED"))
{
config.getRuntime().getStore().updateFields(AssistantProvider.TABLE_NAME, new QRecord().withValue("id", row.getId())
.withValue("healthState", state).withValue("healthCheckedAt", Instant.now().minusSeconds(30)));
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.valueOf(state));
}
for(Instant time : List.of(Instant.now().minusSeconds(600), Instant.now().plusSeconds(30)))
{
config.getRuntime().getStore().updateFields(AssistantProvider.TABLE_NAME, new QRecord().withValue("id", row.getId()).withValue("healthCheckedAt", time));
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
}
config.getRuntime().getStore().updateFields(AssistantProvider.TABLE_NAME, new QRecord().withValue("id", row.getId()).withValue("enabled", false));
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
}
/*******************************************************************************
** Missing adapter/configuration is safe and does not dispatch a completion.
*******************************************************************************/
@Test
void missingAdapterAndSecret() throws Exception
{
AssistantProvider row = row();
var check = check("assistant.provider." + row.getId());
config.withSecretResolver(name -> null);
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.DOWN);
config.getProviderTypes().remove(type.getType());
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
}
/*******************************************************************************
** Successful and failed completions use TEST and preserve manual test history.
*******************************************************************************/
@Test
void modelAccountingAndNoSecrets() throws Exception
{
AssistantModel model = model(row());
Instant previous = Instant.now().minusSeconds(1000);
config.getRuntime().getStore().updateFields(AssistantModel.TABLE_NAME, new QRecord().withValue("id", model.getId())
.withValue("lastTestAt", previous).withValue("lastTestStatus", "OK"));
var check = check("assistant.model." + model.getId());
provider.thenText(SECRET);
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.UP);
assertThat(provider.getRequests().getFirst().getMaxTokens()).isEqualTo(32);
assertThat(probes).isZero();
provider.thenError(ModelProviderException.Kind.NETWORK, SECRET);
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.DOWN);
var usage = config.getRuntime().getStore().queryRecords(AssistantUsage.TABLE_NAME, new QQueryFilter(), 10);
assertThat(usage).hasSize(2).allSatisfy(record -> assertThat(record.getValueString("purpose")).isEqualTo(UsageRecorder.PURPOSE_TEST));
assertThat(config.getRuntime().getStore().listModels().getFirst().getLastTestAt()).isEqualTo(previous);
failCreate = true;
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.DOWN);
}
/*******************************************************************************
** Monthly cap skips even on-demand checks, before any billable request.
*******************************************************************************/
@Test
void modelCostCapAndDisabledProvider() throws Exception
{
AssistantProvider row = row();
AssistantModel model = model(row);
var check = check("assistant.model." + model.getId());
config.getRuntime().getStore().updateFields(AssistantProvider.TABLE_NAME, new QRecord().withValue("id", row.getId()).withValue("monthlyCostCapUsd", BigDecimal.ZERO));
var outcome = check.getCheckInstance().run(new HealthCheckInput().withQInstance(qInstance).withDeadline(Instant.now().plusSeconds(5)).withOnDemand(true));
assertThat(outcome.getStatus()).isEqualTo(HealthStatus.UNKNOWN);
assertThat(outcome.getMessage()).isEqualTo("skipped: cost cap");
assertThat(provider.getRequests()).isEmpty();
assertThat(config.getRuntime().getStore().queryRecords(AssistantUsage.TABLE_NAME, new QQueryFilter(), 10)).isEmpty();
config.getRuntime().getStore().updateFields(AssistantProvider.TABLE_NAME, new QRecord().withValue("id", row.getId()).withValue("enabled", false));
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
}
/*******************************************************************************
** The real runner identity records only its own usage and sees shared caps.
*******************************************************************************/
@Test
void automatedSessionAccountingAndSharedCap() throws Exception
{
AssistantProvider row = row();
AssistantModel model = model(row);
var check = check("assistant.model." + model.getId());
QContext.init(qInstance, new QSession().withUser(new QUser().withIdReference("health_runner"))
.withSecurityKeyValues(Map.of()).withPermissions("health.read", "health.write"));
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.UP);
assertThat(QContext.getQSession().getSecurityKeyValues()).containsOnlyKeys(config.getOwnerSecurityKeyType());
assertThat(QContext.getQSession().hasPermission("assistant.admin")).isFalse();
var usage = config.getRuntime().getStore().queryRecords(AssistantUsage.TABLE_NAME, new QQueryFilter(), 10);
assertThat(usage).hasSize(1);
assertThat(usage.getFirst().getValueString("userId")).isEqualTo("health_runner");
var healthSession = QContext.getQSession();
QContext.init(qInstance, AssistantTestFixture.ownerSession(AssistantTestFixture.ADMIN, config));
config.getRuntime().getStore().insert(new AssistantUsage().withUserId(AssistantTestFixture.EDITOR).withProviderId(row.getId())
.withCostMicros(1_000_000L).withCreateDate(Instant.now()).withPurpose("TURN").withStatus("OK"));
config.getRuntime().getStore().updateFields(AssistantProvider.TABLE_NAME, new QRecord().withValue("id", row.getId()).withValue("monthlyCostCapUsd", BigDecimal.ONE));
QContext.init(qInstance, healthSession);
assertThat(config.getRuntime().getProviderService().monthToDateCostMicros(row.getId())).isZero();
assertThat(run(check).getMessage()).isEqualTo("skipped: cost cap");
assertThat(provider.getRequests()).hasSize(1);
}
/*******************************************************************************
** The kill switch is read fresh before every scheduled model call.
*******************************************************************************/
@Test
void modelKillSwitch() throws Exception
{
var check = check("assistant.model." + model(row()).getId());
config.getRuntime().getStore().insert(new AssistantPolicy().withName("Stop").withKillSwitch(true));
assertThat(run(check).getMessage()).isEqualTo("skipped: kill switch");
assertThat(provider.getRequests()).isEmpty();
}
/*******************************************************************************
** Embedding models reuse ConnectionTest's embedding path, never chat.
*******************************************************************************/
@Test
void embeddingAccountingAndNoSecrets() throws Exception
{
AssistantProvider row = row();
AssistantModel model = config.getRuntime().getStore().insert(new AssistantModel().withProviderId(row.getId()).withModelId("embedding")
.withModelKind("EMBEDDING").withEnabled(true));
var check = check("assistant.model." + model.getId());
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.UP);
failEmbedding = true;
assertThat(run(check).getStatus()).isEqualTo(HealthStatus.DOWN);
assertThat(embeddings).isEqualTo(2);
assertThat(provider.getRequests()).isEmpty();
assertThat(config.getRuntime().getStore().queryRecords(AssistantUsage.TABLE_NAME, new QQueryFilter(), 10))
.hasSize(2).allSatisfy(record -> assertThat(record.getValueString("purpose")).isEqualTo("TEST"));
}
/*******************************************************************************
** Expired input performs no work; a stuck existing probe is interrupted.
*******************************************************************************/
@Test
void providerDeadline() throws Exception
{
var check = check("assistant.provider." + row().getId());
assertThat(check.getCheckInstance().run(new HealthCheckInput().withQInstance(qInstance).withDeadline(Instant.now().minusSeconds(1))).getStatus()).isEqualTo(HealthStatus.DOWN);
assertThat(probes).isZero();
probeWait = new CountDownLatch(1);
assertTimeoutPreemptively(Duration.ofSeconds(3), () ->
{
QContext.init(qInstance, AssistantTestFixture.sessionFor(AssistantTestFixture.ADMIN));
try
{
var outcome = check.getCheckInstance().run(new HealthCheckInput().withQInstance(qInstance).withDeadline(Instant.now().plusMillis(200)));
assertThat(outcome.getMessage()).isEqualTo("Health check timed out");
assertThat(interrupted.await(1, TimeUnit.SECONDS)).isTrue();
}
finally
{
QContext.clear();
}
});
}
/*******************************************************************************
** A slow completion cannot outlive the caller's deadline.
*******************************************************************************/
@Test
void modelDeadline() throws Exception
{
var check = check("assistant.model." + model(row()).getId());
provider.withDelay(Duration.ofSeconds(10)).thenText("OK");
assertTimeoutPreemptively(Duration.ofSeconds(3), () ->
{
QContext.init(qInstance, AssistantTestFixture.sessionFor(AssistantTestFixture.ADMIN));
try
{
var outcome = check.getCheckInstance().run(new HealthCheckInput().withQInstance(qInstance).withDeadline(Instant.now().plusMillis(300)));
assertThat(outcome.getStatus()).isEqualTo(HealthStatus.DOWN);
}
finally
{
QContext.clear();
}
});
}
/*******************************************************************************
** Drive the real runner cache without its scheduler or health-table writes.
*******************************************************************************/
@SuppressWarnings("unchecked")
private HealthRunner cached(String name, HealthCheckOutcome outcome, Instant at) throws Exception
{
HealthRunner runner = new HealthRunner(qInstance, "test", Clock.fixed(at, ZoneOffset.UTC));
var cache = HealthRunner.class.getDeclaredMethod("cacheOutcome", String.class, HealthCheckOutcome.class);
cache.setAccessible(true);
cache.invoke(runner, name, outcome);
var runners = HealthRunner.class.getDeclaredField("RUNNERS");
runners.setAccessible(true);
((Map<QInstance, HealthRunner>) runners.get(null)).put(qInstance, runner);
return (runner);
}
/*******************************************************************************
** Fresh central results avoid duplicate local probes; stale/unknown fall back.
*******************************************************************************/
@Test
void breakerUsesCentralResults() throws Exception
{
AssistantProvider row = row();
var active = new ProviderService.Active(row, type.getType(), "model", provider);
ProviderHealth breaker = new ProviderHealth(config.getRuntime());
String name = "assistant.provider." + row.getId();
for(HealthStatus status : List.of(HealthStatus.UP, HealthStatus.DEGRADED, HealthStatus.DOWN))
{
try(var central = cached(name, new HealthCheckOutcome().withStatus(status), Instant.now().minusSeconds(1)))
{
if(status == HealthStatus.DOWN)
{
assertThatThrownBy(() -> breaker.beforeCall(active)).isInstanceOf(ModelProviderException.class);
}
else
{
breaker.beforeCall(active);
}
assertThat(probes).isZero();
}
}
try(var central = cached(name, HealthCheckOutcome.down("Down"), Instant.now().minusSeconds(60)))
{
breaker.beforeCall(active);
assertThat(probes).isEqualTo(1);
}
config.getRuntime().getStore().updateFields(AssistantProvider.TABLE_NAME, new QRecord().withValue("id", row.getId()).withValue("healthCheckedAt", null));
try(var central = cached(name, HealthCheckOutcome.unknown("Unsupported"), Instant.now()))
{
breaker.beforeCall(active);
assertThat(probes).isEqualTo(2);
}
}
/*******************************************************************************
** The replacement card is generic and links to filtered central health.
*******************************************************************************/
@Test
void centralHealthCard() throws Exception
{
AssistantProvider row = row();
assertThat(qInstance.getWidget(AssistantHealthWidget.NAME).getType()).isEqualTo(WidgetType.STATISTICS.getType());
var renderer = new AssistantHealthWidget();
StatisticsData data = (StatisticsData) renderer.render(new RenderWidgetInput()).getWidgetData();
assertThat(data.getCount()).isEqualTo("UNKNOWN");
assertThat(data.getCountURL()).startsWith("/app/").contains("health_check", "assistant");
try(var central = cached("assistant.provider." + row.getId(), HealthCheckOutcome.up("Reachable"), Instant.now()))
{
data = (StatisticsData) renderer.render(new RenderWidgetInput()).getWidgetData();
assertThat(data.getCount()).isEqualTo("UP");
}
}
}
@@ -0,0 +1,513 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.assistant.jobs;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.kingsrook.qbits.assistant.AssistantRuntime;
import com.kingsrook.qbits.assistant.api.AssistantApi;
import com.kingsrook.qbits.assistant.engine.RunningTurn;
import com.kingsrook.qbits.assistant.engine.TurnEvent;
import com.kingsrook.qbits.assistant.model.AssistantConversation;
import com.kingsrook.qbits.assistant.model.AssistantTurn;
import com.kingsrook.qbits.assistant.provider.fake.ScriptedModelProvider;
import com.kingsrook.qbits.assistant.spi.ModelEventListener;
import com.kingsrook.qbits.assistant.spi.ModelPart;
import com.kingsrook.qbits.assistant.spi.ModelProviderException;
import com.kingsrook.qbits.assistant.spi.ModelRequest;
import com.kingsrook.qbits.assistant.spi.ModelResponse;
import com.kingsrook.qbits.assistant.spi.StreamHandle;
import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
import com.kingsrook.qbits.assistant.uiapi.CoreUiApiGateway;
import com.kingsrook.qbits.assistant.uiapi.UiApiCall;
import com.kingsrook.qbits.assistant.uiapi.UiApiResult;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.instances.QInstanceEnricher;
import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaData;
import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaDataProvider;
import com.kingsrook.qqq.backend.core.jobs.model.JobStatus;
import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgressSnapshot;
import com.kingsrook.qqq.backend.core.jobs.runtime.DurableJobsRuntime;
import com.kingsrook.qqq.backend.core.jobs.store.DurableJobStore;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.utils.JsonUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import static com.kingsrook.qbits.assistant.testing.AssistantTestFixture.EDITOR;
import static com.kingsrook.qqq.backend.core.jobs.DurableJobsTables.TABLE_JOB;
import static com.kingsrook.qqq.backend.core.jobs.DurableJobsTables.TABLE_LOG;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** Real Jobs tray rows, progress and cancellation for assistant answers.
*******************************************************************************/
class JobTrackingRunTrackerTest extends BaseAssistantTest
{
private static final String PROMPT = "PRIVATE-PROMPT-JOBS-CANARY";
private static final String ANSWER = "PRIVATE-ANSWER-JOBS-CANARY";
private final CountDownLatch toolEntered = new CountDownLatch(1);
private final CountDownLatch toolRelease = new CountDownLatch(1);
private DurableJobStore jobs;
private AssistantApi api;
private ControlledProvider controlled;
/*******************************************************************************
** Add jobs only to this fixture, leaving runtime startup explicit per test.
*******************************************************************************/
@BeforeEach
void initializeJobs() throws Exception
{
new DurableJobsMetaDataProvider().defineAll(qInstance, AssistantTestFixture.BACKEND, null);
DurableJobsMetaData settings = new DurableJobsMetaData().withWorkerThreads(1).withLeaseTtlSeconds(3)
.withReaperIntervalSeconds(1).withPollIntervalMillis(50).withPodId("test-pod");
qInstance.withSupplementalMetaData(settings);
new QInstanceEnricher(qInstance).enrich();
jobs = new DurableJobStore(settings);
config.getRuntime().close();
config.withUiApiGateway(new CoreUiApiGateway(config::getUiApiBaseUrl, qInstance)
{
/*******************************************************************************
** Hold the real count call so the tool phase can be inspected in storage.
*******************************************************************************/
@Override
public UiApiResult invoke(UiApiCall call) throws QException
{
if("/table/{tableName}/count".equals(call.getPathTemplate()))
{
toolEntered.countDown();
try
{
assertThat(toolRelease.await(15, TimeUnit.SECONDS)).as("tool released").isTrue();
}
catch(InterruptedException e)
{
Thread.currentThread().interrupt();
throw (new QException("Test tool interrupted", e));
}
}
return (super.invoke(call));
}
});
config.setRuntime(new AssistantRuntime(qInstance, config));
var session = AssistantTestFixture.ownerSession(EDITOR, config);
QContext.init(qInstance, session);
api = new AssistantApi(config.getRuntime(), session);
assertThat(config.getRunTracker()).isExactlyInstanceOf(JobTrackingRunTracker.class);
}
/*******************************************************************************
** Release every gate before draining both runtimes, even after an assertion.
*******************************************************************************/
@AfterEach
void stopJobs()
{
toolRelease.countDown();
if(controlled != null)
{
controlled.handles.forEach(handle -> handle.release.countDown());
}
config.getRuntime().close();
DurableJobsRuntime.stop(qInstance, Duration.ofSeconds(5));
}
/*******************************************************************************
** A single linked owner job progresses through thinking, tool and writing.
*******************************************************************************/
@Test
void answerTracksProgressAndCompletes() throws Exception
{
assertThat(DurableJobsRuntime.start(qInstance)).isNotNull();
ControlledHandle thinking = new ControlledHandle(new ModelResponse().withParts(List.of(
ModelPart.toolUse("count-1", "count_records", input("table", "blog_post", "filter", filter("AND")))))
.withStopReason(ModelResponse.STOP_TOOL_USE), null, false);
ControlledHandle writing = answer(false);
useProvider(thinking, writing);
Integer conversationId = (Integer) api.createConversation(Map.of()).get("id");
RunningTurn run = api.startTurn(conversationId, Map.of("text", PROMPT));
entered(thinking);
QRecord job = onlyJob();
Integer jobId = job.getValueInteger("id");
assertThat(job.getValueString("processName")).isEqualTo("assistant.turn");
assertThat(job.getValueString("label")).isEqualTo("Assistant answer");
assertThat(job.getValueString("userId")).isEqualTo(EDITOR);
assertThat(job.getValueString("tableName")).isEqualTo(AssistantConversation.TABLE_NAME);
assertThat(JsonUtils.toObject(job.getValueString("sourceJson"), Map.class)).containsEntry("recordId", conversationId);
assertPhase(jobId, "thinking", null);
thinking.release.countDown();
assertThat(toolEntered.await(15, TimeUnit.SECONDS)).as("tool entered").isTrue();
assertPhase(jobId, "tool", "Counting records");
toolRelease.countDown();
entered(writing);
assertPhase(jobId, "writing", null);
writing.release.countDown();
assertThat(completed(run).getStatus()).isEqualTo("COMPLETE");
job = onlyJob();
assertThat(job.getValueInteger("id")).isEqualTo(jobId);
assertThat(job.getValueString("status")).isEqualTo("COMPLETED");
assertThat(job.getValueString("progressMessage")).isEqualTo("Answered.");
assertThat(job.getValueString("statusReason")).isEqualTo("Answered.");
assertThat(run.getLog().after(0, 1)).filteredOn(event -> TurnEvent.TOOL_RESULT.equals(event.getType()))
.singleElement().satisfies(event -> assertThat(event.getData()).containsEntry("outcome", "OK"));
assertPrivateTextAbsent();
}
/*******************************************************************************
** The runtime's periodic tick observes the same row change as Jobs tray Stop.
*******************************************************************************/
@Test
void jobsTrayStopCancelsAnswer() throws Exception
{
DurableJobsRuntime.start(qInstance);
ControlledHandle writing = answer(false);
useProvider(writing);
RunningTurn run = startTurn();
entered(writing);
Integer jobId = onlyJob().getValueInteger("id");
jobs.updateJob(jobId, Map.of("status", "CANCELLING"));
assertThat(completed(run).getStatus()).isEqualTo("CANCELLED");
assertThat(writing.isCancelled()).isTrue();
assertThat(onlyJob().getValueString("status")).isEqualTo("CANCELLED");
assertThat(onlyJob().getValueString("progressMessage")).isEqualTo("Stopped.");
assertPrivateTextAbsent();
}
/*******************************************************************************
** Even partial output and provider errors containing private text stay out.
*******************************************************************************/
@Test
void failedAnswerHasOnlyFixedJobMessages() throws Exception
{
DurableJobsRuntime.start(qInstance);
ControlledHandle writing = answer(true);
useProvider(writing);
RunningTurn run = startTurn();
entered(writing);
writing.release.countDown();
assertThat(completed(run).getStatus()).isEqualTo("ERROR");
QRecord job = onlyJob();
assertThat(job.getValueString("status")).isEqualTo("FAILED");
assertThat(job.getValueString("progressMessage")).isEqualTo("The answer did not finish.");
assertThat(job.getValueString("statusReason")).isEqualTo("The answer did not finish.");
assertThat(jobs.query(TABLE_LOG, new QQueryFilter())).isNotEmpty();
assertPrivateTextAbsent();
}
/*******************************************************************************
** Hosts without jobs keep ordinary answers and safe null tracker callbacks.
*******************************************************************************/
@Test
void absentRuntimeDoesNotTrackOrDisruptAnswer() throws Exception
{
assertThat(DurableJobsRuntime.get(qInstance)).isNull();
AssistantRunTracker tracker = config.getRunTracker();
assertThat(tracker.start(EDITOR, 1, 1, "ignored")).isNull();
tracker.progress(null, "thinking", PROMPT);
tracker.onCancel(null, () ->
{
throw (new AssertionError("No job to cancel"));
});
tracker.finish(null, "COMPLETE", ANSWER);
provider.thenText(ANSWER);
RunningTurn run = startTurn();
assertThat(completed(run).getStatus()).isEqualTo("COMPLETE");
assertThat(jobs.query(TABLE_JOB, new QQueryFilter())).isEmpty();
assertThat(jobs.query(TABLE_LOG, new QQueryFilter())).isEmpty();
}
/*******************************************************************************
** Core mutates phase objects; every new job must start with fresh phases.
*******************************************************************************/
@Test
void jobsHaveIndependentPhaseState() throws Exception
{
DurableJobsRuntime.start(qInstance);
AssistantRunTracker tracker = config.getRunTracker();
Integer conversationId = (Integer) api.createConversation(Map.of()).get("id");
String first = tracker.start(EDITOR, 1, conversationId, PROMPT);
tracker.progress(first, "writing", ANSWER);
String second = tracker.start(EDITOR, 2, conversationId, PROMPT);
QRecord secondJob = jobs.getJob(Integer.valueOf(second));
JobProgressSnapshot snapshot = JsonUtils.toObject(secondJob.getValueString("progressJson"), JobProgressSnapshot.class);
assertThat(snapshot.getPhases()).extracting(JobPhase::getStatus)
.containsExactly(JobPhase.Status.PENDING, JobPhase.Status.PENDING, JobPhase.Status.PENDING);
tracker.progress(second, "thinking", PROMPT);
tracker.progress(first, "unknown", PROMPT);
tracker.progress("missing", "tool", PROMPT);
assertPhase(Integer.valueOf(first), "writing", null);
assertPhase(Integer.valueOf(second), "thinking", null);
tracker.finish(first, "COMPLETE", ANSWER);
tracker.finish(second, "COMPLETE", ANSWER);
assertPrivateTextAbsent();
}
/*******************************************************************************
** Terminal mapping is fixed, including unknown and absent assistant statuses.
*******************************************************************************/
@ParameterizedTest
@CsvSource({ "COMPLETE,COMPLETED,Answered.", "CANCELLED,CANCELLED,Stopped.",
"ERROR,FAILED,The answer did not finish.", "unknown,FAILED,The answer did not finish.",
",FAILED,The answer did not finish." })
void terminalMapping(String status, JobStatus expected, String message)
{
assertThat(JobTrackingRunTracker.jobStatus(status)).isEqualTo(expected);
assertThat(JobTrackingRunTracker.finishMessage(status)).isEqualTo(message);
}
/*******************************************************************************
** Use independently controlled model calls without changing the tracker.
*******************************************************************************/
private void useProvider(ControlledHandle... handles)
{
controlled = new ControlledProvider(List.of(handles));
config.withModelProvider(controlled);
}
/*******************************************************************************
** Partial text is emitted before this response is released or fails.
*******************************************************************************/
private ControlledHandle answer(boolean fail)
{
return (new ControlledHandle(new ModelResponse().withParts(List.of(ModelPart.text(ANSWER)))
.withStopReason(ModelResponse.STOP_END_TURN), ANSWER, fail));
}
/*******************************************************************************
** Create a conversation and answer as the fixture's editor.
*******************************************************************************/
private RunningTurn startTurn() throws Exception
{
Integer conversationId = (Integer) api.createConversation(Map.of()).get("id");
return (api.startTurn(conversationId, Map.of("text", PROMPT)));
}
/*******************************************************************************
** Wait until a model call has emitted its optional text and is blocked.
*******************************************************************************/
private void entered(ControlledHandle handle) throws InterruptedException
{
assertThat(handle.entered.await(15, TimeUnit.SECONDS)).as("model call entered").isTrue();
}
/*******************************************************************************
** Check persisted progress while the answer is still running.
*******************************************************************************/
private void assertPhase(Integer jobId, String phase, String message) throws QException
{
QRecord job = jobs.getJob(jobId);
assertThat(job.getValueString("status")).isEqualTo("RUNNING");
assertThat(job.getValueString("currentPhase")).isEqualTo(phase);
assertThat(job.getValueString("progressMessage")).isEqualTo(message);
}
/*******************************************************************************
** Assert uniqueness across all jobs, not just those matching this turn.
*******************************************************************************/
private QRecord onlyJob() throws QException
{
List<QRecord> rows = jobs.query(TABLE_JOB, new QQueryFilter());
assertThat(rows).hasSize(1);
return (rows.getFirst());
}
/*******************************************************************************
** Inspect every persisted job and log field for both private canaries.
*******************************************************************************/
private void assertPrivateTextAbsent() throws QException
{
for(String table : List.of(TABLE_JOB, TABLE_LOG))
{
assertThat(jobs.query(table, new QQueryFilter())).allSatisfy(row ->
assertThat(JsonUtils.toJson(row)).doesNotContain(PROMPT, ANSWER));
}
}
/*******************************************************************************
** Wait for finalization and cleanup before reading the assistant turn.
*******************************************************************************/
private AssistantTurn completed(RunningTurn run) throws Exception
{
await(() -> config.getRuntime().getRunningTurn(run.getTurnId()) == null);
assertThat(run.getLog().isClosed()).isTrue();
return (config.getRuntime().getStore().getTurn(EDITOR, run.getTurnId()));
}
/*******************************************************************************
** Bound polling by observable completion rather than fixed timing guesses.
*******************************************************************************/
private static void await(Callable<Boolean> condition) throws Exception
{
long deadline = System.nanoTime() + Duration.ofSeconds(15).toNanos();
while(!condition.call() && System.nanoTime() < deadline)
{
Thread.sleep(10);
}
assertThat(condition.call()).as("condition reached within fifteen seconds").isTrue();
}
/*******************************************************************************
** Script model responses with separate gates for thinking and writing.
*******************************************************************************/
private static class ControlledProvider extends ScriptedModelProvider
{
private final List<ControlledHandle> handles;
private final AtomicInteger next = new AtomicInteger();
/*******************************************************************************
** Retain the scripted handles for teardown and assertions.
*******************************************************************************/
ControlledProvider(List<ControlledHandle> handles)
{
this.handles = handles;
}
/*******************************************************************************
** Publish partial text, then let the engine wait on the selected handle.
*******************************************************************************/
@Override
public StreamHandle stream(ModelRequest request, ModelEventListener listener)
{
ControlledHandle handle = handles.get(next.getAndIncrement());
if(handle.text != null)
{
listener.onTextDelta(handle.text);
}
return (handle);
}
}
/*******************************************************************************
** A bounded, cancellable model call that can fail after streaming text.
*******************************************************************************/
private static class ControlledHandle implements StreamHandle
{
private final ModelResponse response;
private final String text;
private final boolean fail;
private final CountDownLatch entered = new CountDownLatch(1);
private final CountDownLatch release = new CountDownLatch(1);
private volatile boolean cancelled;
/*******************************************************************************
** Set the eventual response and optional partial-output failure.
*******************************************************************************/
ControlledHandle(ModelResponse response, String text, boolean fail)
{
this.response = response;
this.text = text;
this.fail = fail;
}
/*******************************************************************************
** Stop unblocks the engine without test intervention.
*******************************************************************************/
@Override
public void cancel()
{
cancelled = true;
release.countDown();
}
/*******************************************************************************
** Report whether the real cancellation callback reached the provider.
*******************************************************************************/
@Override
public boolean isCancelled()
{
return (cancelled);
}
/*******************************************************************************
** Wait until the test or the Jobs tray releases this model call.
*******************************************************************************/
@Override
public ModelResponse await(Duration timeout) throws ModelProviderException, InterruptedException
{
entered.countDown();
if(!release.await(timeout.toMillis(), TimeUnit.MILLISECONDS))
{
throw (new ModelProviderException(ModelProviderException.Kind.TIMEOUT, null, "Test provider timed out"));
}
if(cancelled)
{
return (new ModelResponse().withStopReason(ModelResponse.STOP_CANCELLED));
}
if(fail)
{
throw (new ModelProviderException(ModelProviderException.Kind.OTHER, null, PROMPT + " " + ANSWER));
}
return (response);
}
}
}
@@ -561,7 +561,7 @@ class AssistantSettingsTest extends BaseAssistantTest
assertThat(data.getMonthCostMicros()).isZero();
String html = render();
assertThat(html).contains("Answering").contains("Fake").contains("In use").contains("Last test passed").contains("…wxyz")
assertThat(html).contains("Configuration and usage").contains("Fake").contains("In use").doesNotContain("Answering", "Last test passed").contains("…wxyz")
.contains("of $50.00 cap").contains("list_tables").contains("Model call #").contains("25% of 4 requests").doesNotContain(KEY).doesNotContain("<table");
}