QRun-IO/qbit-assistant
0
0 Fork 0
Code Issues 3 Pull Requests 7 Actions Packages Projects Releases 23 Wiki Activity

Chat turns are not jobs; link bulk loads to their job; notification audit #63

Merged
KofTwentyTwo merged 5 commits from feature/turns-not-jobs into develop 2026-10-19 13:44:44 +00:00
Conversation 0 Commits 5 Files Changed 14
+128 -1026
14 changed files with 128 additions and 1026 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CHANGELOG.md
+7
View File
@@ -1,5 +1,12 @@
# Changelog
## Unreleased - Chat turns are not jobs
- Ordinary chat turns no longer create core tracking jobs: they stay out of the Jobs tray and Jobs tables, and core no longer sends an "Assistant answer finished" notification for every answer. Crash safety is unchanged; it comes from the assistant's own turn table (leased, heartbeated `assistant_turn` rows, seq-numbered `assistant_turn_event` replay and the abandoned-turn sweep). The panel's Stop (`POST /turns/{id}/cancel`) is unchanged.
- Removed the `AssistantRunTracker` SPI, `JobTrackingRunTracker` and `AssistantQBitConfig.withRunTracker` / `setRunTracker` / `getRunTracker` (see UPGRADE.md).
- Real multi-step work stays a real job: a confirmed bulk load still launches through `POST /qqq/v1/processes/{p}/jobs`, and its tool result now carries `jobHref` (the job's `/app` link) next to `jobId`, plus a record ref to the job; the confirm response's `href` links to the job.
- Scheduled task notifications (output mode Notification or Both) now go through core `QAlerts.notify` (category Job, the owner as the only audience, a link to the conversation, success/error/warning severity by run status), replacing the reflective call into an older alerts SPI that no longer matched.
## Unreleased - Attachments
- Files attached in the panel (`pageContext.attachments`, core upload refs) are read for the question: as the signed-in user's own upload, in memory, with per-file and per-question byte and character caps (`AssistantQBitConfig.attachments`). The type is sniffed from content; executables, archives and unknown binary data are refused. Text, Markdown and JSON are decoded with charset detection, PDF page text is extracted (PDFBox via QQQ core, no scripts, host font mapper preserved), CSV/TSV/XLSX become a header + rows + count preview, and PNG/JPEG/GIF/WebP images go to vision models as image parts.
UPGRADE.md
+20
View File
@@ -1,5 +1,25 @@
# Upgrading qbit-assistant
## Chat turns are not jobs (Unreleased)
Ordinary chat turns no longer create core tracking jobs, so they no longer appear
in the Jobs tray or the Jobs tables and no longer trigger a per-answer "finished"
notification. No schema change and no configuration is needed.
The `AssistantRunTracker` SPI and `JobTrackingRunTracker` are removed, along with
`AssistantQBitConfig.withRunTracker`, `setRunTracker` and `getRunTracker`. A host
that called any of them must delete the call: `withRunTracker(AssistantRunTracker.NOOP)`
is now the only behavior, and a custom tracker has no replacement hook. Turn
progress, Stop and recovery are available through the assistant's own routes
(`GET /turns/active`, the turn events stream and `POST /turns/{id}/cancel`).
Bulk loads the assistant launches are still durable jobs. Their confirmed tool
result now includes `jobHref` (the job's in-app link) and the confirm response's
`href` points at the job.
Scheduled task notifications now use core `QAlerts.notify`; they reach whatever
alerts sink the host installed (or core's logging sink when none is).
## Provider setup state (Unreleased)
Apply the **Provider setup state** block at the end of `docs/schema-postgres.sql`
qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/AssistantQBitConfig.java
-35
View File
@@ -32,8 +32,6 @@ import java.util.regex.Pattern;
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;
@@ -107,7 +105,6 @@ public class AssistantQBitConfig implements QBitConfig
private Integer shutdownGraceSeconds = 8;
private String defaultConcurrencyMode = "SUPERSEDE";
private Boolean registerShutdownHook = true;
private AssistantRunTracker runTracker = new JobTrackingRunTracker();
private Integer maxTurnSeconds = 300;
private Integer toolCallTimeoutSeconds = 60;
@@ -2361,36 +2358,4 @@ public class AssistantQBitConfig implements QBitConfig
return (this);
}
/*******************************************************************************
** Getter for runTracker
*******************************************************************************/
@JsonIgnore
public AssistantRunTracker getRunTracker()
{
return (this.runTracker == null ? AssistantRunTracker.NOOP : this.runTracker);
}
/*******************************************************************************
** Setter for runTracker
*******************************************************************************/
public void setRunTracker(AssistantRunTracker runTracker)
{
this.runTracker = runTracker;
}
/*******************************************************************************
** Fluent setter for runTracker
*******************************************************************************/
public AssistantQBitConfig withRunTracker(AssistantRunTracker runTracker)
{
this.runTracker = runTracker;
return (this);
}
}
qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java
+7 -15
View File
@@ -480,21 +480,6 @@ public class AssistantApi
RunningTurn runningTurn = new RunningTurn(turn.getId(), userId);
TurnEngine engine = new TurnEngine(runtime, session, userId, conversation, userMessage, assistantMessage, turn, text, pageContext, policy, runningTurn,
budget.softExceeded() ? budget.message() : null).withCredentials(credentials).withTranscript(transcript);
try
{
if(config.getRunTracker() != null)
{
///////////////////////////////////////////////////////////////////
// never the conversation title: it is derived from the prompt //
///////////////////////////////////////////////////////////////////
runningTurn.setTrackingId(config.getRunTracker().start(userId, turn.getId(), conversationId, "Assistant answer"));
config.getRunTracker().onCancel(runningTurn.getTrackingId(), () -> runningTurn.cancel("USER"));
}
}
catch(Exception e)
{
LOG.warn("Assistant run tracker start failed", logPair("turnId", turn.getId()));
}
runtime.register(runningTurn);
if(!runningTurn.isFinalized())
{
@@ -919,6 +904,13 @@ public class AssistantApi
{
rs.put("href", outcome.getAuditHref());
}
else if(!result.getRecordRefs().isEmpty() && TurnEngine.isAppHref(result.getRecordRefs().get(0).getHref()))
{
//////////////////////////////////////////////////////////////
// e.g. the job a confirmed bulk load launched //
//////////////////////////////////////////////////////////////
rs.put("href", result.getRecordRefs().get(0).getHref());
}
return (rs);
}
qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/RunningTurn.java
-21
View File
@@ -40,7 +40,6 @@ public class RunningTurn
private volatile Thread engineThread;
private volatile Consumer<String> forceFinalizer;
private volatile long cancelledNanos;
private volatile String trackingId;
private volatile StreamHandle currentHandle;
@@ -264,24 +263,4 @@ public class RunningTurn
return (isCancelled() && cancelledNanos != 0 && System.nanoTime() - cancelledNanos >= seconds * 1_000_000_000L);
}
/*******************************************************************************
** Getter for trackingId
*******************************************************************************/
public String getTrackingId()
{
return (trackingId);
}
/*******************************************************************************
** Setter for trackingId
*******************************************************************************/
public void setTrackingId(String trackingId)
{
this.trackingId = trackingId;
}
}
qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/TurnEngine.java
+14 -24
View File
@@ -792,8 +792,8 @@ public class TurnEngine implements Runnable
}
///////////////////////////////////////////////////////////////////////
// providers stream on their own (virtual) threads, which have no //
// QContext: run the delta handler in the turn's context, or progress //
// writes (JobLogWriter) fail with no instance //
// QContext: run the delta handler in the turn's context, so phase //
// and text events are recorded with the turn's instance and session //
///////////////////////////////////////////////////////////////////////
CapturedContext turnContext = QContext.capture();
ModelEventListener listener = new ModelEventListener()
@@ -1418,30 +1418,21 @@ public class TurnEngine implements Runnable
runtime.getScheduledTurn().detach(conversation.getId(), session.getUuid());
}
});
String finalStatus = status;
String finalText = errorText;
try
runtime.unregister(turn.getId());
runtime.getRateLimiter().release(userId);
if(transcript != null && !engineRunning)
{
safely(() -> config.getRunTracker().finish(running.getTrackingId(), finalStatus, finalText));
transcript.close();
}
finally
else
{
runtime.unregister(turn.getId());
runtime.getRateLimiter().release(userId);
if(transcript != null && !engineRunning)
{
transcript.close();
}
else
{
// A wedged provider can still log after forced finalization. The
// engine's finally releases its log protection when it returns.
TranscriptRedactor.detach();
}
if(previousTranscript != null && previousTranscript != transcript)
{
previousTranscript.attach();
}
// A wedged provider can still log after forced finalization. The
// engine's finally releases its log protection when it returns.
TranscriptRedactor.detach();
}
if(previousTranscript != null && previousTranscript != transcript)
{
previousTranscript.attach();
}
}
}
@@ -1491,7 +1482,6 @@ public class TurnEngine implements Runnable
running.setPhase(phase, toolName);
String label = toolName == null ? phase : ToolLabels.running(toolName);
emit(TurnEvent.TURN_PHASE, map("phase", phase, "toolName", toolName, "label", label));
safely(() -> config.getRunTracker().progress(running.getTrackingId(), phase, label));
}
qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/AssistantRunTracker.java
-80
View File
@@ -1,80 +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.jobs;
import com.kingsrook.qqq.backend.core.exceptions.QException;
/*******************************************************************************
** 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
{
AssistantRunTracker NOOP = new AssistantRunTracker()
{
/*******************************************************************************
** No tracking is configured.
*******************************************************************************/
@Override
public String start(String userId, Integer turnId, Integer conversationId, String label)
{
return (null);
}
/*******************************************************************************
** No tracking is configured.
*******************************************************************************/
@Override
public void progress(String trackingId, String phase, String message)
{
}
/*******************************************************************************
** No tracking is configured.
*******************************************************************************/
@Override
public void finish(String trackingId, String status, String message)
{
}
};
/*******************************************************************************
** Publish a run and return its tracking id.
*******************************************************************************/
String start(String userId, Integer turnId, Integer conversationId, String label) throws QException;
/*******************************************************************************
** Publish a changed phase.
*******************************************************************************/
void progress(String trackingId, String phase, String message);
/*******************************************************************************
** Publish the terminal outcome.
*******************************************************************************/
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) throws QException
{
}
}
qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTracker.java
-170
View File
@@ -1,170 +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.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.");
}
}
qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/scheduled/ScheduledAlerts.java
+37 -27
View File
@@ -21,14 +21,24 @@
package com.kingsrook.qbits.assistant.scheduled;
import com.kingsrook.qbits.assistant.model.AssistantConversation;
import com.kingsrook.qbits.assistant.model.AssistantScheduledRun;
import com.kingsrook.qqq.backend.core.alerts.AlertLink;
import com.kingsrook.qqq.backend.core.alerts.NotificationAudience;
import com.kingsrook.qqq.backend.core.alerts.NotificationCategory;
import com.kingsrook.qqq.backend.core.alerts.NotificationSeverity;
import com.kingsrook.qqq.backend.core.alerts.NotificationSignal;
import com.kingsrook.qqq.backend.core.alerts.QAlerts;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
/*******************************************************************************
** Optional Alerts notification SPI bridge. The audience is exactly one user;
** notification bodies contain only status and a link, never answer text.
** The one notification a scheduled task's owner asked for (output mode
** NOTIFICATION or BOTH): through core's non-throwing QAlerts.notify, so it
** reaches whatever alerts sink the host installed (or core's logging sink).
** The audience is exactly the task's owner; the body has only the status and
** the link, never answer text.
*******************************************************************************/
public class ScheduledAlerts
{
@@ -37,36 +47,36 @@ public class ScheduledAlerts
/*******************************************************************************
** Reflection keeps the optional notification implementation out of the ABI.
** Notify the run's owner. Never throws.
*******************************************************************************/
public void raise(AssistantScheduledRun run)
{
LOG.info("Scheduled assistant result", logPair("runId", run.getId()), logPair("status", run.getStatus()));
try
QAlerts.notify(signal(run));
}
/*******************************************************************************
** The notification for a finished run.
*******************************************************************************/
static NotificationSignal signal(AssistantScheduledRun run)
{
String status = run.getStatus() == null ? "FINISHED" : run.getStatus();
NotificationSignal signal = new NotificationSignal()
.withTitle("OK".equals(status) ? "Scheduled assistant task finished" : "Scheduled assistant task did not finish")
.withBody("Status: " + status + ". See the conversation.")
.withSeverity("OK".equals(status) ? NotificationSeverity.SUCCESS : "FAILED".equals(status) ? NotificationSeverity.ERROR : NotificationSeverity.WARNING)
.withCategory(NotificationCategory.JOB)
.withSourceName("assistant")
.withSourceKey("assistant/scheduled/run/" + run.getId())
.withAudience(NotificationAudience.user(run.getUserId()));
if(run.getConversationId() != null)
{
String prefix = "com.kingsrook.qbits.alerts.spi.";
Class<?> type = Class.forName(prefix + "NotificationSignal");
Object signal = type.getConstructor(String.class).newInstance("Scheduled assistant task finished");
type.getMethod("withSourceName", String.class).invoke(signal, "assistant");
type.getMethod("withSourceKey", String.class).invoke(signal, "assistant/scheduled/run/" + run.getId());
type.getMethod("withBody", String.class).invoke(signal, "Status: " + run.getStatus() + ". See conversation.");
Class<?> audienceType = Class.forName(prefix + "NotificationAudience");
Object audience = audienceType.getMethod("user", String.class).invoke(null, run.getUserId());
type.getMethod("withAudience", audienceType).invoke(signal, audience);
if(run.getConversationId() != null)
{
Class<?> linkType = Class.forName(prefix + "AlertLink");
Object link = linkType.getMethod("toPath", String.class, String.class).invoke(null,
"/app/assistantAdmin/assistant_conversation/" + run.getConversationId(), "See conversation");
type.getMethod("withLink", linkType).invoke(signal, link);
}
Object runtime = Class.forName("com.kingsrook.qbits.alerts.AlertsRuntime").getMethod("get").invoke(null);
Object sink = runtime.getClass().getMethod("sink").invoke(runtime);
sink.getClass().getMethod("notify", type).invoke(sink, signal);
}
catch(Exception | LinkageError e)
{
LOG.info("Scheduled assistant notification unavailable", logPair("runId", run.getId()));
signal.withSourceTableName(AssistantConversation.TABLE_NAME)
.withSourceRecordId(String.valueOf(run.getConversationId()))
.withLink(AlertLink.toRecord(AssistantConversation.TABLE_NAME, run.getConversationId(), "Open the conversation"));
}
return (signal);
}
}
qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadTools.java
+23 -1
View File
@@ -30,12 +30,15 @@ import com.kingsrook.qbits.assistant.spi.ModelToolDefinition;
import com.kingsrook.qbits.assistant.tools.UiApiToolbox;
import com.kingsrook.qbits.assistant.uiapi.UiApiCall;
import com.kingsrook.qbits.assistant.uiapi.UiApiResult;
import com.kingsrook.qqq.backend.core.actions.ai.AiRecordRef;
import com.kingsrook.qqq.backend.core.actions.ai.AiToolOutcome;
import com.kingsrook.qqq.backend.core.actions.ai.AiToolResult;
import com.kingsrook.qqq.backend.core.actions.ai.output.AiDataEnvelope;
import com.kingsrook.qqq.backend.core.actions.ai.output.AiLinkBuilder;
import com.kingsrook.qqq.backend.core.actions.ai.tools.AiJsonSchema;
import com.kingsrook.qqq.backend.core.actions.ai.tools.AiToolException;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.DurableJobsTables;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.utils.JsonUtils;
import com.kingsrook.qqq.backend.core.utils.StringUtils;
@@ -537,7 +540,9 @@ public class BulkLoadTools
/*******************************************************************************
** Confirmed run: through the runner seam.
** Confirmed run: through the runner seam. A launched load is a real job
** (visible in the Jobs tray and tables); its id and in-app link go into the
** result and its record refs, so the conversation links to the job.
*******************************************************************************/
private void run(Call call, AiToolResult result, UiApiToolbox.Context context, String tableName, Map<String, Object> values) throws QException
{
@@ -549,6 +554,13 @@ public class BulkLoadTools
Map<String, Object> out = new LinkedHashMap<>();
out.put("status", handle.status());
out.put("jobId", handle.jobId());
if(handle.jobId() != null)
{
String jobHref = jobHref(handle.jobId());
out.put("jobHref", jobHref);
result.getRecordRefs().add(new AiRecordRef().withTableName(DurableJobsTables.TABLE_JOB).withTableLabel("Job").withId(handle.jobId())
.withLabel("Job " + handle.jobId()).withHref(jobHref));
}
out.put("processUUID", handle.processUUID());
out.put("summary", handle.summary());
result.withOutcome(ok ? AiToolOutcome.OK : AiToolOutcome.ERROR).withSummary(shortText(handle.summary())).withPermissionCheck(call.permissionCheck())
@@ -563,6 +575,16 @@ public class BulkLoadTools
/*******************************************************************************
** The in-app link to a durable job's record.
*******************************************************************************/
public static String jobHref(Long jobId)
{
return (AiLinkBuilder.recordHref(DurableJobsTables.TABLE_JOB, jobId));
}
/*******************************************************************************
** A tool's input or result as kept in the tool-call log: profiles and
** counts, never sample values, example bullets or value mappings.
qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/DurableTurnsTest.java
-140
View File
@@ -32,7 +32,6 @@ import com.kingsrook.qbits.assistant.AssistantQBitConfig;
import com.kingsrook.qbits.assistant.AssistantRuntime;
import com.kingsrook.qbits.assistant.api.AssistantApi;
import com.kingsrook.qbits.assistant.api.AssistantApiException;
import com.kingsrook.qbits.assistant.jobs.AssistantRunTracker;
import com.kingsrook.qbits.assistant.model.AssistantConversation;
import com.kingsrook.qbits.assistant.model.AssistantMessage;
import com.kingsrook.qbits.assistant.model.AssistantTurn;
@@ -208,42 +207,6 @@ class DurableTurnsTest extends BaseAssistantTest
/*******************************************************************************
** Streamed text arrives on the provider's own thread; the "writing" progress
** it triggers still runs in the turn's QContext (the job tracker writes
** through QQQ, which needs the instance).
*******************************************************************************/
@Test
void streamedProgressRunsInTheTurnContext() throws Exception
{
List<String> contexts = new CopyOnWriteArrayList<>();
config.setRunTracker(new AssistantRunTracker()
{
@Override
public String start(String userId, Integer turnId, Integer conversationId, String label)
{
return ("job-" + turnId);
}
@Override
public void progress(String trackingId, String phase, String message)
{
contexts.add(phase + ":" + (QContext.getQInstance() != null));
}
@Override
public void finish(String trackingId, String status, String message)
{
}
});
provider.thenText("A streamed answer on the provider's thread");
AssistantTurn row = completed(api.startTurn(conversation(), Map.of("text", "hello")));
assertThat(row.getStatus()).isEqualTo("COMPLETE");
assertThat(contexts).contains("writing:true").doesNotContain("writing:false");
}
/*******************************************************************************
** A throwing post-processing collaborator cannot strand a streamed answer.
*******************************************************************************/
@@ -360,8 +323,6 @@ class DurableTurnsTest extends BaseAssistantTest
{
ControlledProvider controlled = controlled(true);
controlled.emitText = false;
RecordingTracker tracker = new RecordingTracker();
config.setRunTracker(tracker);
config.setMaxTurnSeconds(1);
RunningTurn run = api.startTurn(conversation(), Map.of("text", "wedged"));
try
@@ -373,12 +334,10 @@ class DurableTurnsTest extends BaseAssistantTest
assertThat(row.getCancelReason()).isEqualTo("TIMEOUT");
assertThat(store.getMessage(EDITOR, row.getAssistantMessageId()).getPartsJson()).contains(TurnEngine.TIMEOUT_TEXT);
assertThat(controlled.handles.getFirst().cancelled).isTrue();
assertThat(tracker.finishes.get()).isEqualTo(1);
int events = run.getLog().after(0, 1).size();
controlled.release();
await(() -> controlled.handles.getFirst().returned);
assertThat(run.getLog().after(0, 1)).hasSize(events);
assertThat(tracker.finishes.get()).isEqualTo(1);
}
finally
{
@@ -576,8 +535,6 @@ class DurableTurnsTest extends BaseAssistantTest
List<String> errors = new ArrayList<>();
config.validate(qInstance, errors);
assertThat(errors).contains("turnHeartbeatSeconds must be less than turnLeaseSeconds.", "defaultConcurrencyMode must be SUPERSEDE or CONCURRENT.");
config.withRunTracker(null);
assertThat(config.getRunTracker()).isSameAs(AssistantRunTracker.NOOP);
}
@@ -621,8 +578,6 @@ class DurableTurnsTest extends BaseAssistantTest
void rollingDeploy() throws Exception
{
ControlledProvider controlled = controlled(true);
RecordingTracker tracker = new RecordingTracker();
config.setRunTracker(tracker);
Integer conversation = conversation();
RunningTurn run = api.startTurn(conversation, Map.of("text", "wait"));
try
@@ -634,7 +589,6 @@ class DurableTurnsTest extends BaseAssistantTest
assertThat(row.getStatus()).isEqualTo("ERROR");
assertThat(row.getErrorCode()).isEqualTo("INTERRUPTED");
assertThat(store.getMessage(EDITOR, row.getAssistantMessageId()).getPartsJson()).contains(TurnEngine.RESTART_TEXT);
assertThat(tracker.finishes.get()).isEqualTo(1);
int messages = store.listMessages(EDITOR, conversation, null, 100).size();
assertThatThrownBy(() -> api.startTurn(conversation, Map.of("text", "retry")))
.isInstanceOfSatisfying(AssistantApiException.class, e ->
@@ -668,35 +622,6 @@ class DurableTurnsTest extends BaseAssistantTest
/*******************************************************************************
** Tracker progress is phase-based; its Stop callback cancels the local run.
*******************************************************************************/
@Test
void trackerAndTrackerFailures() throws Exception
{
ControlledProvider controlled = controlled(false);
RecordingTracker tracker = new RecordingTracker();
config.setRunTracker(tracker);
RunningTurn run = api.startTurn(conversation(), Map.of("text", "wait"));
controlled.awaitCalls(1);
await(() -> tracker.cancel != null);
tracker.cancel.run();
assertThat(completed(run).getStatus()).isEqualTo("CANCELLED");
assertThat(tracker.starts.get()).isEqualTo(1);
assertThat(tracker.phases).containsExactly("thinking", "writing");
assertThat(tracker.finishes.get()).isEqualTo(1);
assertThat(tracker.label).isEqualTo("Assistant answer");
tracker.fail = true;
config.withModelProvider(provider);
config.getRuntime().getProviderService().invalidate();
provider.thenText("Still works");
RunningTurn throwing = api.startTurn(conversation(), Map.of("text", "hello"));
assertThat(completed(throwing).getStatus()).isEqualTo("COMPLETE");
assertThat(tracker.finishes.get()).isEqualTo(2);
}
/*******************************************************************************
** Build a remote persisted turn with an assistant message to recover.
*******************************************************************************/
@@ -953,69 +878,4 @@ class DurableTurnsTest extends BaseAssistantTest
.withStopReason(cancelled ? ModelResponse.STOP_CANCELLED : ModelResponse.STOP_END_TURN));
}
}
/*******************************************************************************
** Records the host job-tray contract and can throw from every callback.
*******************************************************************************/
private static class RecordingTracker implements AssistantRunTracker
{
private final AtomicInteger starts = new AtomicInteger();
private final AtomicInteger finishes = new AtomicInteger();
private final List<String> phases = new CopyOnWriteArrayList<>();
private volatile Runnable cancel;
private String label;
private boolean fail;
/*******************************************************************************
** Record publication.
*******************************************************************************/
@Override
public String start(String userId, Integer turnId, Integer conversationId, String label)
{
starts.incrementAndGet();
this.label = label;
if(fail)
{
throw (new IllegalStateException("tracker start failed"));
}
return ("job-" + turnId);
}
/*******************************************************************************
** Record a phase transition.
*******************************************************************************/
@Override
public void progress(String trackingId, String phase, String message)
{
phases.add(phase);
if(fail)
{
throw (new IllegalStateException("tracker progress failed"));
}
}
/*******************************************************************************
** Record terminal notification exactly once.
*******************************************************************************/
@Override
public void finish(String trackingId, String status, String message)
{
finishes.incrementAndGet();
if(fail)
{
throw (new IllegalStateException("tracker finish failed"));
}
}
/*******************************************************************************
** Retain the host's Stop callback.
*******************************************************************************/
@Override
public void onCancel(String trackingId, Runnable cancel)
{
this.cancel = cancel;
}
}
}
qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTrackerTest.java
-513
View File
@@ -1,513 +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.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);
}
}
}
qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/scheduled/ScheduledTasksTest.java
+11
View File
@@ -60,6 +60,8 @@ import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
import com.kingsrook.qqq.backend.core.actions.ai.AiFieldMetaData;
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
import com.kingsrook.qqq.backend.core.alerts.NotificationSeverity;
import com.kingsrook.qqq.backend.core.alerts.NotificationSignal;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
@@ -421,6 +423,15 @@ class ScheduledTasksTest extends BaseAssistantTest
assertThat(failed.getStatus()).isEqualTo("FAILED");
assertThat(failed.getError()).doesNotContain("Private provider detail");
new ScheduledAlerts().raise(failed);
NotificationSignal signal = ScheduledAlerts.signal(failed);
assertThat(signal.getAudience().getValues()).containsExactly(AssistantTestFixture.EDITOR);
assertThat(signal.getSeverity()).isEqualTo(NotificationSeverity.ERROR);
assertThat(signal.getBody()).contains("FAILED").doesNotContain("Private provider detail");
assertThat(signal.getLink().getTableName()).isEqualTo("assistant_conversation");
assertThat(signal.getLink().getRecordId()).isEqualTo(failed.getConversationId());
assertThat(ScheduledAlerts.signal(new AssistantScheduledRun().withId(1).withUserId(AssistantTestFixture.EDITOR).withStatus("OK")).getSeverity())
.isEqualTo(NotificationSeverity.SUCCESS);
}
qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadToolsTest.java
+9
View File
@@ -22,6 +22,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import com.fasterxml.jackson.core.type.TypeReference;
import com.kingsrook.qbits.assistant.AssistantQBitConfig;
import com.kingsrook.qbits.assistant.access.DataAccessMode;
import com.kingsrook.qbits.assistant.api.AssistantApi;
@@ -336,6 +337,14 @@ class BulkLoadToolsTest extends BaseAssistantTest
assertThat(runs.get(0).savedBulkLoadProfileId()).isEqualTo(7);
assertThat(runs.get(0).uploadRef()).isEqualTo("up_1");
assertThat(runs.get(0).tableName()).isEqualTo("blog_post");
//////////////////////////////////////////////////////////////////
// the load is a real job: the result carries its id and link, //
// and the confirmation links the conversation to it //
//////////////////////////////////////////////////////////////////
assertThat(ran.get("href")).isEqualTo("/app/jobs_job/42");
Map<String, Object> message = JsonUtils.toObject(String.valueOf(ran.get("message")), new TypeReference<Map<String, Object>>() {});
assertThat(message).containsEntry("jobId", 42).containsEntry("jobHref", "/app/jobs_job/42");
}