From ac2107bd978bf55c51b3feaa3c33170287f74ca0 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 16 Oct 2026 15:29:48 -0500 Subject: [PATCH 1/4] feat(assistant)!: chat turns are not jobs; remove the run tracker SPI Ordinary chat turns no longer create core tracking jobs, so they stay out of the Jobs tray and tables and no longer trigger a per-answer "finished" notification. Crash safety comes from the assistant's own turn table (leased, heartbeated assistant_turn rows, seq-numbered event replay and the abandoned-turn sweep), which is unchanged. Removes AssistantRunTracker, JobTrackingRunTracker, the config accessors and the tracking id on RunningTurn, with their tests. The panel's Stop (POST /turns/{id}/cancel) is unchanged and still covered. Co-Authored-By: Claude Opus 5.5 --- .../qbits/assistant/AssistantQBitConfig.java | 35 -- .../qbits/assistant/api/AssistantApi.java | 15 - .../qbits/assistant/engine/RunningTurn.java | 21 - .../qbits/assistant/engine/TurnEngine.java | 38 +- .../assistant/jobs/AssistantRunTracker.java | 80 --- .../assistant/jobs/JobTrackingRunTracker.java | 170 ------ .../assistant/engine/DurableTurnsTest.java | 140 ----- .../jobs/JobTrackingRunTrackerTest.java | 513 ------------------ 8 files changed, 14 insertions(+), 998 deletions(-) delete mode 100644 qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/AssistantRunTracker.java delete mode 100644 qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTracker.java delete mode 100644 qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTrackerTest.java diff --git a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/AssistantQBitConfig.java b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/AssistantQBitConfig.java index 83adb66..cd1b5d0 100644 --- a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/AssistantQBitConfig.java +++ b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/AssistantQBitConfig.java @@ -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; @@ -2321,36 +2318,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); - } - } diff --git a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java index ece2c1d..c0219ee 100644 --- a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java +++ b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java @@ -469,21 +469,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()) { diff --git a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/RunningTurn.java b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/RunningTurn.java index 4d6ba75..53cadba 100644 --- a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/RunningTurn.java +++ b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/RunningTurn.java @@ -40,7 +40,6 @@ public class RunningTurn private volatile Thread engineThread; private volatile Consumer 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; - } - } diff --git a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/TurnEngine.java b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/TurnEngine.java index 38a264e..780ef2d 100644 --- a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/TurnEngine.java +++ b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/TurnEngine.java @@ -775,8 +775,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() @@ -1396,30 +1396,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(); } } } @@ -1469,7 +1460,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)); } diff --git a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/AssistantRunTracker.java b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/AssistantRunTracker.java deleted file mode 100644 index 2fe36c1..0000000 --- a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/AssistantRunTracker.java +++ /dev/null @@ -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 - { - } -} diff --git a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTracker.java b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTracker.java deleted file mode 100644 index 73aed7b..0000000 --- a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTracker.java +++ /dev/null @@ -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 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."); - } - -} diff --git a/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/DurableTurnsTest.java b/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/DurableTurnsTest.java index 6b49a61..2677d01 100644 --- a/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/DurableTurnsTest.java +++ b/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/DurableTurnsTest.java @@ -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 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 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 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; - } - } } diff --git a/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTrackerTest.java b/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTrackerTest.java deleted file mode 100644 index a991f83..0000000 --- a/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/jobs/JobTrackingRunTrackerTest.java +++ /dev/null @@ -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 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 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 handles; - private final AtomicInteger next = new AtomicInteger(); - - /******************************************************************************* - ** Retain the scripted handles for teardown and assertions. - *******************************************************************************/ - ControlledProvider(List 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); - } - } -} -- 2.54.0 From f7235690b63d07a46753ffb4a5f93cf96362500a Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 16 Oct 2026 15:29:51 -0500 Subject: [PATCH 2/4] feat(assistant): link a launched bulk load to its job A confirmed bulk load still launches a real durable job through POST /qqq/v1/processes/{p}/jobs. Its tool result now carries jobHref (the job's /app link) next to jobId and a record ref to the job, and the confirm response's href points at the job, so the conversation links to it. Co-Authored-By: Claude Opus 5.5 --- .../qbits/assistant/api/AssistantApi.java | 7 ++++++ .../tools/bulkload/BulkLoadTools.java | 24 ++++++++++++++++++- .../tools/bulkload/BulkLoadToolsTest.java | 9 +++++++ 3 files changed, 39 insertions(+), 1 deletion(-) diff --git a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java index c0219ee..28173b6 100644 --- a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java +++ b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java @@ -893,6 +893,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); } diff --git a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadTools.java b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadTools.java index 91401cc..ab63a55 100644 --- a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadTools.java +++ b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadTools.java @@ -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 values) throws QException { @@ -549,6 +554,13 @@ public class BulkLoadTools Map 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. diff --git a/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadToolsTest.java b/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadToolsTest.java index 824735f..b63838e 100644 --- a/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadToolsTest.java +++ b/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadToolsTest.java @@ -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 message = JsonUtils.toObject(String.valueOf(ran.get("message")), new TypeReference>() {}); + assertThat(message).containsEntry("jobId", 42).containsEntry("jobHref", "/app/jobs_job/42"); } -- 2.54.0 From e1f64cfa05745d0f79e8a5f8ff0fea6a0ac3d84c Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 16 Oct 2026 15:29:53 -0500 Subject: [PATCH 3/4] fix(assistant): send scheduled task notifications through core QAlerts The notification a task owner asked for (output mode Notification or Both) now goes through core QAlerts.notify: category Job, the owner as the only audience, severity by run status and a link to the conversation. It replaces a reflective call into an older alerts SPI that no longer matched. Co-Authored-By: Claude Opus 5.5 --- .../assistant/scheduled/ScheduledAlerts.java | 64 +++++++++++-------- .../scheduled/ScheduledTasksTest.java | 11 ++++ 2 files changed, 48 insertions(+), 27 deletions(-) diff --git a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/scheduled/ScheduledAlerts.java b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/scheduled/ScheduledAlerts.java index 08d18a6..5122366 100644 --- a/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/scheduled/ScheduledAlerts.java +++ b/qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/scheduled/ScheduledAlerts.java @@ -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); } } diff --git a/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/scheduled/ScheduledTasksTest.java b/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/scheduled/ScheduledTasksTest.java index 89b2917..1113a0d 100644 --- a/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/scheduled/ScheduledTasksTest.java +++ b/qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/scheduled/ScheduledTasksTest.java @@ -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); } -- 2.54.0 From fc5026594d84cfa85d7096cce40fc1c7a2285976 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 16 Oct 2026 15:29:54 -0500 Subject: [PATCH 4/4] docs: changelog and upgrade notes for turns that are not jobs Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 +++++++ UPGRADE.md | 20 ++++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 688af47..ab9c81c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 - Optional provider health - An enabled provider that is not the default and not the only route for any configured routing (an enabled profile's simple/complex model, a budget downgrade model or a knowledge source's embedding model, with no fallback on another usable provider) is optional: when it is unreachable, its `assistant.provider.` and `assistant.model.` checks report DEGRADED with a message naming the provider requests use instead, so one unfinished local server no longer turns the whole `assistant` component DOWN. The default provider and sole routes still report DOWN. A failed provider probe still carries `reachable=false`, so the circuit breaker keeps refusing the provider. diff --git a/UPGRADE.md b/UPGRADE.md index b8d2206..1f76866 100644 --- a/UPGRADE.md +++ b/UPGRADE.md @@ -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` -- 2.54.0