Compare commits
14
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
079e8d39e2
|
||
|
|
ef933d6bea | ||
|
|
385b61019b
|
||
|
|
a7184d3be9
|
||
|
|
2a3691c0ea
|
||
|
|
b6982eff9f
|
||
|
|
1012b4b497 | ||
|
|
76f9b286c6
|
||
|
|
9b4708fafc | ||
|
|
15938e07e1
|
||
|
|
b2c97d1000 | ||
|
|
ae8e394a7b | ||
|
|
e8beb7444a
|
||
|
|
6c36917beb
|
No files matched your search
+20
-1
@@ -1,5 +1,25 @@
|
||||
# Changelog
|
||||
|
||||
## Unreleased - Jobs tray
|
||||
|
||||
- Track assistant answers in the core Jobs tray through `JobTracking`, with thinking, tool and writing progress and Stop support.
|
||||
- Require QQQ 4.1.0-RC.8 and the host's durable jobs runtime and jobs tables; answers remain untracked when the runtime is absent.
|
||||
- Default to `JobTrackingRunTracker`; opt out with `withRunTracker(AssistantRunTracker.NOOP)`.
|
||||
|
||||
## Unreleased - In-app help
|
||||
|
||||
- Ship the host-neutral Assistant help bundle through QQQ's native registry, replacing the temporary loader.
|
||||
- Cover Assistant tables, fields, actions, widgets and selectors; retain live metadata references and add a bundle coverage gate.
|
||||
- Document HOST bundle overrides and remove deployment-specific provider, classification, retention and backup guidance.
|
||||
|
||||
## Unreleased - Transcript redaction
|
||||
|
||||
- Tokenize classified PII/PHI in stored messages, tool inputs/results, replay events, pending actions and derived summaries using QQQ secured-data tokens. Preserve eligible model requests and live SSE.
|
||||
- Stage content until turn completion, redact split deltas and classify criteria even when no records match; protect assistant logs with a scoped core redactor.
|
||||
- Reject proposals whose parameters were redacted; prevent confirmation before turn completion.
|
||||
- Read core table-level classification metadata and add secured-data canary tests for storage, audits, logs and in-memory model access.
|
||||
- Preserve durable-turn watchdog, supersede, shutdown and restart recovery with turn-scoped transcript redaction; widen token-bearing columns and entity limits, with H2-safe upgrade SQL.
|
||||
- Require QQQ 4.1.0-RC.7 (qqq#1182); retain the default audit-history opt-in and avoid duplicated routing headers against current core. RC.7 is released.
|
||||
## Unreleased - Foreign keys
|
||||
|
||||
- Reference DDL: add the missing foreign keys (conversation tree, usage, data audit,
|
||||
@@ -26,7 +46,6 @@
|
||||
`message.start.startedAt`, cancelReason `SUPERSEDED`. The 429 response stays
|
||||
documented for budgets and per-minute limits; the "answer already in progress"
|
||||
refusal is no longer returned. Cancel keeps 202 documented for older replicas.
|
||||
||||||| befb90a
|
||||
|
||||
## Unreleased - Starter defaults
|
||||
|
||||
|
||||
@@ -52,6 +52,8 @@ Always use the wrapper: Maven 3.10 cannot read the parent POM. The committed def
|
||||
|
||||
## Invariants (do not break)
|
||||
|
||||
- **Transcript redaction:** stored transcripts never contain raw PII/PHI values; tokens via `ClassifiedValueTokens`. Redact storage copies before truncation/encryption; eligible model inputs and live SSE remain unchanged.
|
||||
|
||||
- **Sessions:**
|
||||
- Scheduled runs log in as the designated user through `POST /qqq/v1/manageSession`; resolve only its returned UUID, verify the user id, and never construct or elevate a session. The QQQ-supplied scheduler context touches only assistant bookkeeping.
|
||||
- Never use `QSystemUserSession` or construct a `QSession` in main code (ArchUnit).
|
||||
|
||||
@@ -30,6 +30,25 @@ Hosts register providers through `AssistantQBitConfig` in `qbit-assistant`.
|
||||
|
||||
Data tools use the QQQ UI API (`/qqq/v1`) with the signed-in user's session. The UI API enforces permissions and record locks; assistant data-access policy and field exposure can only narrow access. Only record-free navigation and help use the core AI tool kernel. Every tool result reaches the model wrapped as untrusted data.
|
||||
|
||||
## Transcript privacy
|
||||
|
||||
QQQ **4.1.0-RC.7** supplies the secured-data tokens and log hook (`qqq#1182`).
|
||||
Stored transcripts never contain the raw PII/PHI values read by a turn or supplied
|
||||
in classified tool criteria/fields. Messages, tool logs, replay events, knowledge
|
||||
snippets and pending payloads store `ClassifiedValueTokens` tokens; scheduled
|
||||
summaries use the redacted message. The model and live user stream retain the
|
||||
values permitted by PHI routing. History and replay show tokens instead.
|
||||
|
||||
Content is finalized at turn completion; interrupted servers leave value-free
|
||||
placeholders. Proposals requiring redacted parameters are rejected and must be
|
||||
completed in the application. Existing transcripts are not migrated. See
|
||||
[transcript redaction](docs/DESIGN.md#transcript-redaction-secured-data-15).
|
||||
The dependency upgrade stays draft until RC.7 is published; local verification:
|
||||
|
||||
```powershell
|
||||
.\mvnw.cmd -q verify "-Dqqq.snapshot.version=4.1.0-RC7PRE-SNAPSHOT"
|
||||
```
|
||||
|
||||
## Host audit attribution (X1/Y1)
|
||||
|
||||
Use QQQ **4.1.0-RC.6 or later**. The QBit explicitly denies every assistant
|
||||
@@ -706,3 +725,18 @@ remain untouched, as do explicit object and permission-rule denials.
|
||||
History visibility uses core's `isDenied` read/all-operation check. RC.6 has no
|
||||
operation argument; the later write-only deny patterns must remain separate
|
||||
from that check so they do not hide the read tool.
|
||||
|
||||
## In-app help
|
||||
|
||||
The QBit registers [`help/qbit-assistant.yaml`](qbit-assistant/src/main/resources/help/qbit-assistant.yaml)
|
||||
with QQQ's help-bundle registry during production. Hosts with help content storage use QQQ's
|
||||
startup bundle sync; registration is harmless when the host has no `helpContent` table.
|
||||
|
||||
To customize the wording, register a separate bundle with `layer: HOST` using
|
||||
`HelpBundles.register(qInstance, "help/my-application.yaml")`. Entries with the same key and
|
||||
role override this QBIT bundle. User-edited help remains protected by core's sync.
|
||||
Keep live `{{...}}` references so labels and metadata values follow the host's configuration.
|
||||
|
||||
`AssistantTablesGuardrailTest.everyOwnedTargetHasBundleHelp` checks coverage and live references
|
||||
against the produced test instance with `HelpBundleAssertions`; it runs in the normal Maven build.
|
||||
Update the bundle's literal version alongside the artifact version when releasing.
|
||||
+20
-7
@@ -1,5 +1,13 @@
|
||||
# Upgrading qbit-assistant
|
||||
|
||||
## Transcript redaction (Unreleased)
|
||||
|
||||
Apply the transcript-redaction column widening block in `docs/schema-postgres.sql`
|
||||
to existing databases. Stop reasons, request IDs, tool-call IDs, feedback/data-audit/
|
||||
budget-override reasons and scheduled-run summaries/errors now use `TEXT`; token
|
||||
expansion must not fail a write or truncate secured-data tokens. The corresponding
|
||||
entity field limits are removed, including pending-action and tool-result summaries.
|
||||
|
||||
## Foreign keys (2026-10-14)
|
||||
|
||||
Recommended: apply the "Foreign keys" block at the end of
|
||||
@@ -61,13 +69,18 @@ reattach through the existing events route. Consumers should recognize
|
||||
`turn.phase` and `message.start.startedAt`. Starts during shutdown return
|
||||
503 `SHUTTING_DOWN`; clients may retry against a ready replica.
|
||||
|
||||
Hosts may wire `AssistantRunTracker` for Jobs tray publication, progress, Stop,
|
||||
and terminal notification. It is a no-op by default and all callbacks are
|
||||
isolated from turn execution. Tracking ids are local to the owning runtime;
|
||||
reaping an already dead owner does not invent another tracking job. Hosts should
|
||||
associate their durable tracking row with the supplied turn id. The SPI is a
|
||||
stand-in for QQQ #1210.
|
||||
||||||| befb90a
|
||||
`JobTrackingRunTracker` is now the default `AssistantRunTracker`, using core
|
||||
`JobTracking` in QQQ **4.1.0-RC.8** to publish answers in the Jobs tray. The host
|
||||
must provide the jobs tables (through qbit-jobs / core `DurableJobsMetaDataProvider`)
|
||||
and run `DurableJobsRuntime`. Without that runtime, answers complete normally
|
||||
and are simply not tracked.
|
||||
|
||||
Each job is owned by the assistant user, links to its `assistant_conversation`,
|
||||
and reports thinking, tool and writing phases. Tool progress contains only the
|
||||
tool label; terminal messages are fixed and never contain prompts or answers.
|
||||
The tray's Stop cancels the assistant turn. All tracker callbacks remain isolated
|
||||
from turn execution. Opt out with `withRunTracker(AssistantRunTracker.NOOP)`;
|
||||
hosts may still supply their own `AssistantRunTracker`.
|
||||
|
||||
## Starter defaults, setup card and table help
|
||||
|
||||
|
||||
@@ -637,3 +637,32 @@ The shared golden task asks "How many images do I have by type?": five UI count
|
||||
calls establish 81 JPEG, 20 PNG and zero other configured types, then a presentation
|
||||
with two rows, 80.2%/19.8% shares, two pie points and working MIME filter links.
|
||||
The same task is included in the opt-in `live` Anthropic eval suite.
|
||||
|
||||
|
||||
## Transcript redaction (secured data §15)
|
||||
|
||||
`TranscriptRedactor` collects classified values from tools using `ClassificationSource`,
|
||||
including core table/field metadata. Stored record fields use
|
||||
`ClassifiedValueTokens.forRecordField`; criteria/proposals without a record use
|
||||
`forContext`. Core selects `shred:v1`, `hmac:v1`, or `changed:v1`. Knowledge snippets
|
||||
retain their source field's token. Matching is case-sensitive and includes canonical,
|
||||
displayed, escaped, encoded and explicitly truncated variants.
|
||||
|
||||
`AssistantStore` stages content in turn memory and writes placeholders until
|
||||
finalization, protecting backend writes and their audits. It redacts before truncation
|
||||
and encryption. This covers messages, tool/bulk logs, replay events, pending payloads
|
||||
and audit reasons; scheduled summaries derive from redacted messages. Model requests
|
||||
and live SSE remain unchanged under PHI routing. Persisted text deltas are joined
|
||||
before redaction, with the text in the first delta and remaining deltas empty;
|
||||
sequence numbers remain stable.
|
||||
|
||||
Crashes or unfinished tools at timeout leave placeholders. Proposals remain
|
||||
non-executable (`PREPARING`) until finalization; redacted proposals are rejected and
|
||||
must be completed in the application. A core `LogRedactorInterface` hook covers all
|
||||
threads until the turn and its tool workers finish, then releases the raw values.
|
||||
|
||||
History stays redacted; later turns re-read records. Existing transcripts are not
|
||||
migrated, and arbitrary text is not inferred to be classified without a matching
|
||||
observed value or classified criterion/field. Canary tests scan every assistant table,
|
||||
core audits and captured logs, while proving eligible models receive original values.
|
||||
Requires QQQ 4.1.0-RC.7 (`qqq#1182`); keep the PR draft until publication.
|
||||
@@ -49,7 +49,7 @@ CREATE TABLE assistant_turn
|
||||
status VARCHAR(16),
|
||||
classification VARCHAR(16),
|
||||
classification_tables VARCHAR(500),
|
||||
stop_reason VARCHAR(64),
|
||||
stop_reason TEXT,
|
||||
error_code VARCHAR(32),
|
||||
cancel_reason VARCHAR(64),
|
||||
model VARCHAR(200),
|
||||
@@ -88,7 +88,7 @@ CREATE TABLE assistant_tool_call
|
||||
turn_id INTEGER,
|
||||
conversation_id INTEGER,
|
||||
user_id VARCHAR(255),
|
||||
tool_call_id VARCHAR(128),
|
||||
tool_call_id TEXT,
|
||||
tool_name VARCHAR(64),
|
||||
input_json TEXT,
|
||||
outcome VARCHAR(16),
|
||||
@@ -118,7 +118,7 @@ CREATE TABLE assistant_feedback
|
||||
conversation_id INTEGER,
|
||||
user_id VARCHAR(255),
|
||||
rating INTEGER,
|
||||
reason VARCHAR(64),
|
||||
reason TEXT,
|
||||
comment TEXT,
|
||||
create_date TIMESTAMP WITH TIME ZONE
|
||||
);
|
||||
@@ -267,7 +267,7 @@ CREATE TABLE assistant_usage
|
||||
cache_read_tokens BIGINT,
|
||||
cache_write_tokens BIGINT,
|
||||
cost_micros BIGINT,
|
||||
request_id VARCHAR(100),
|
||||
request_id TEXT,
|
||||
create_date TIMESTAMP WITH TIME ZONE
|
||||
);
|
||||
CREATE INDEX assistant_usage_create_date ON assistant_usage (create_date);
|
||||
@@ -310,7 +310,7 @@ CREATE TABLE assistant_pending_action
|
||||
conversation_id INTEGER,
|
||||
turn_id INTEGER,
|
||||
user_id VARCHAR(255),
|
||||
tool_call_id VARCHAR(128),
|
||||
tool_call_id TEXT,
|
||||
tool_name VARCHAR(64),
|
||||
ui_api_operation VARCHAR(255),
|
||||
table_name VARCHAR(128),
|
||||
@@ -382,7 +382,7 @@ CREATE TABLE assistant_data_audit
|
||||
classification VARCHAR(16),
|
||||
tables VARCHAR(500),
|
||||
decision VARCHAR(16),
|
||||
reason VARCHAR(1000),
|
||||
reason TEXT,
|
||||
create_date TIMESTAMP WITH TIME ZONE
|
||||
);
|
||||
CREATE INDEX assistant_data_audit_user_id ON assistant_data_audit (user_id);
|
||||
@@ -415,7 +415,7 @@ CREATE TABLE assistant_budget_override
|
||||
scope_value VARCHAR(200),
|
||||
extra_value NUMERIC,
|
||||
unlimited_until TIMESTAMP WITH TIME ZONE,
|
||||
reason VARCHAR(1000) NOT NULL,
|
||||
reason TEXT NOT NULL,
|
||||
granted_by VARCHAR(200),
|
||||
expires_at TIMESTAMP WITH TIME ZONE NOT NULL,
|
||||
create_date TIMESTAMP WITH TIME ZONE,
|
||||
@@ -566,9 +566,9 @@ CREATE TABLE assistant_scheduled_run (
|
||||
status VARCHAR(20) NOT NULL CHECK (status IN ('RUNNING', 'OK', 'FAILED', 'SKIPPED', 'BLOCKED')),
|
||||
conversation_id INTEGER REFERENCES assistant_conversation(id) ON DELETE SET NULL,
|
||||
turn_id INTEGER REFERENCES assistant_turn(id) ON DELETE SET NULL,
|
||||
summary VARCHAR(2000),
|
||||
summary TEXT,
|
||||
cost_micros BIGINT,
|
||||
error VARCHAR(500),
|
||||
error TEXT,
|
||||
create_date TIMESTAMPTZ,
|
||||
modify_date TIMESTAMPTZ,
|
||||
UNIQUE (task_id, scheduled_for)
|
||||
@@ -701,3 +701,15 @@ BEGIN
|
||||
ALTER TABLE assistant_model ADD CONSTRAINT fk_assistant_model_provider FOREIGN KEY (provider_id) REFERENCES assistant_provider (id) ON DELETE CASCADE;
|
||||
END IF;
|
||||
END $$;
|
||||
|
||||
-- Transcript redaction: tokens can expand beyond the original payload bounds.
|
||||
-- Apply to existing databases as well; SET DATA TYPE is also valid in H2.
|
||||
ALTER TABLE assistant_turn ALTER COLUMN stop_reason SET DATA TYPE TEXT;
|
||||
ALTER TABLE assistant_tool_call ALTER COLUMN tool_call_id SET DATA TYPE TEXT;
|
||||
ALTER TABLE assistant_feedback ALTER COLUMN reason SET DATA TYPE TEXT;
|
||||
ALTER TABLE assistant_usage ALTER COLUMN request_id SET DATA TYPE TEXT;
|
||||
ALTER TABLE assistant_pending_action ALTER COLUMN tool_call_id SET DATA TYPE TEXT;
|
||||
ALTER TABLE assistant_data_audit ALTER COLUMN reason SET DATA TYPE TEXT;
|
||||
ALTER TABLE assistant_budget_override ALTER COLUMN reason SET DATA TYPE TEXT;
|
||||
ALTER TABLE assistant_scheduled_run ALTER COLUMN summary SET DATA TYPE TEXT;
|
||||
ALTER TABLE assistant_scheduled_run ALTER COLUMN error SET DATA TYPE TEXT;
|
||||
@@ -202,7 +202,7 @@
|
||||
</build>
|
||||
|
||||
<profiles>
|
||||
<!-- The assistant needs QQQ 4.1.0-RC.6: the AI tool kernel (qqq#1078), the -->
|
||||
<!-- The assistant needs QQQ 4.1.0-RC.7: the AI tool kernel (qqq#1078), the -->
|
||||
<!-- developer panels (qqq#1076) and the UI API extension point, invoker -->
|
||||
<!-- and credentials (qqq#1128). Until -->
|
||||
<!-- qbit-build-parent imports the 4.1 BOM, this -->
|
||||
@@ -215,7 +215,9 @@
|
||||
<activeByDefault>true</activeByDefault>
|
||||
</activation>
|
||||
<properties>
|
||||
<qqq.snapshot.version>4.1.0-RC.6</qqq.snapshot.version>
|
||||
<!-- RC.7 provides classified-value tokens, log redaction and native help bundles; -->
|
||||
<!-- RC.8 provides core job tracking (JobTracking, the Jobs tray). -->
|
||||
<qqq.snapshot.version>4.1.0-RC.8</qqq.snapshot.version>
|
||||
</properties>
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
|
||||
@@ -114,6 +114,7 @@
|
||||
<directory>src/main/resources</directory>
|
||||
<filtering>true</filtering>
|
||||
<includes>
|
||||
<include>help/*.yaml</include>
|
||||
<include>qbit-assistant.properties</include>
|
||||
</includes>
|
||||
</resource>
|
||||
@@ -121,6 +122,7 @@
|
||||
<directory>src/main/resources</directory>
|
||||
<filtering>false</filtering>
|
||||
<excludes>
|
||||
<exclude>help/*.yaml</exclude>
|
||||
<exclude>qbit-assistant.properties</exclude>
|
||||
</excludes>
|
||||
</resource>
|
||||
|
||||
@@ -33,6 +33,7 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
import com.kingsrook.qbits.assistant.classification.ClassificationSource;
|
||||
import com.kingsrook.qbits.assistant.classification.ConfiguredClassificationSource;
|
||||
import com.kingsrook.qbits.assistant.jobs.AssistantRunTracker;
|
||||
import com.kingsrook.qbits.assistant.jobs.JobTrackingRunTracker;
|
||||
import com.kingsrook.qbits.assistant.rag.VectorStore;
|
||||
import com.kingsrook.qbits.assistant.spi.AssistantModelProvider;
|
||||
import com.kingsrook.qbits.assistant.spi.AssistantProviderType;
|
||||
@@ -106,7 +107,7 @@ public class AssistantQBitConfig implements QBitConfig
|
||||
private Integer shutdownGraceSeconds = 8;
|
||||
private String defaultConcurrencyMode = "SUPERSEDE";
|
||||
private Boolean registerShutdownHook = true;
|
||||
private AssistantRunTracker runTracker = AssistantRunTracker.NOOP;
|
||||
private AssistantRunTracker runTracker = new JobTrackingRunTracker();
|
||||
|
||||
private Integer maxTurnSeconds = 300;
|
||||
private Integer maxUserMessageChars = 8000;
|
||||
|
||||
+7
-2
@@ -26,7 +26,6 @@ import com.kingsrook.qbits.assistant.developer.AssistantDeveloperPanels;
|
||||
import com.kingsrook.qbits.assistant.home.AssistantAccessWidget;
|
||||
import com.kingsrook.qbits.assistant.home.AssistantHome;
|
||||
import com.kingsrook.qbits.assistant.metadata.AssistantChannelWall;
|
||||
import com.kingsrook.qbits.assistant.metadata.AssistantHelp;
|
||||
import com.kingsrook.qbits.assistant.metadata.AssistantInstanceMetaData;
|
||||
import com.kingsrook.qbits.assistant.metadata.AssistantMetaDataBuilder;
|
||||
import com.kingsrook.qbits.assistant.metadata.AssistantRecordScreens;
|
||||
@@ -43,6 +42,7 @@ import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerMultiOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.help.bundles.HelpBundles;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.permissions.ClientChannelsMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
|
||||
@@ -204,13 +204,18 @@ public class AssistantQBitProducer implements QBitMetaDataProducer<AssistantQBit
|
||||
|
||||
QAppMetaData app = builder.buildAdminApp(tables);
|
||||
qInstance.addApp(app);
|
||||
AssistantHelp.apply(qInstance);
|
||||
// Registration is metadata-only; core startup sync guards hosts without helpContent.
|
||||
HelpBundles.register(qInstance, "help/qbit-assistant.yaml");
|
||||
|
||||
AssistantChannelWall.register(qInstance);
|
||||
if(Boolean.TRUE.equals(qBitConfig.getRecordHistory()))
|
||||
{
|
||||
enableRecordHistory(ClientChannelsMetaData.ofOrWithNew(qInstance));
|
||||
}
|
||||
else
|
||||
{
|
||||
ClientChannelsMetaData.ofOrWithNew(qInstance).withDeniedPattern("assistant", "audit*");
|
||||
}
|
||||
postProduceActions(new MetaDataProducerMultiOutput(), qInstance);
|
||||
return (new MetaDataProducerMultiOutput());
|
||||
}
|
||||
|
||||
@@ -48,6 +48,7 @@ import com.kingsrook.qbits.assistant.model.AssistantToolCall;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantTurn;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantTurnEvent;
|
||||
import com.kingsrook.qbits.assistant.persistence.AssistantStore;
|
||||
import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
|
||||
import com.kingsrook.qbits.assistant.tools.UiApiToolbox;
|
||||
import com.kingsrook.qbits.assistant.tools.bulkload.BulkLoadTools;
|
||||
import com.kingsrook.qqq.backend.core.actions.ai.AiToolOutcome;
|
||||
@@ -374,11 +375,13 @@ public class AssistantApi
|
||||
requestStop(remaining.removeFirst(), "SUPERSEDED");
|
||||
}
|
||||
|
||||
TranscriptRedactor transcript = new TranscriptRedactor(runtime.getQInstance(), config.getClassificationSource());
|
||||
transcript.attach();
|
||||
try
|
||||
{
|
||||
if(!StringUtils.hasContent(conversation.getTitle()))
|
||||
{
|
||||
String title = deriveTitle(text);
|
||||
String title = text;
|
||||
store.updateFields(AssistantConversation.TABLE_NAME, new QRecord().withValue("id", conversationId).withValue("title", title));
|
||||
conversation.setTitle(title);
|
||||
}
|
||||
@@ -423,7 +426,7 @@ 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);
|
||||
budget.softExceeded() ? budget.message() : null).withCredentials(credentials).withTranscript(transcript);
|
||||
try
|
||||
{
|
||||
if(config.getRunTracker() != null)
|
||||
@@ -456,9 +459,14 @@ public class AssistantApi
|
||||
}
|
||||
catch(QException | RuntimeException e)
|
||||
{
|
||||
transcript.close();
|
||||
runtime.getRateLimiter().release(userId);
|
||||
throw (e);
|
||||
}
|
||||
finally
|
||||
{
|
||||
TranscriptRedactor.detach();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -659,6 +667,8 @@ public class AssistantApi
|
||||
{
|
||||
return (cancelReason == null ? current : requestStop(current, cancelReason));
|
||||
}
|
||||
// Recovery only appends fixed notices to already redacted stored parts (or
|
||||
// crash placeholders). It never reloads model/tool values or raw user text.
|
||||
String status = cancelReason == null ? "ERROR" : "CANCELLED";
|
||||
String text = cancelReason == null ? "This answer was interrupted (the server restarted). Ask again."
|
||||
: ("SUPERSEDED".equals(cancelReason) ? TurnEngine.SUPERSEDED_TEXT : "Stopped by you.");
|
||||
@@ -770,6 +780,12 @@ public class AssistantApi
|
||||
throw (new AssistantApiException(409, "This action is already " + action.getStatus().toLowerCase() + "."));
|
||||
}
|
||||
|
||||
AssistantTurn proposingTurn = store.getTurn(userId, action.getTurnId());
|
||||
if(proposingTurn != null && isRunning(proposingTurn))
|
||||
{
|
||||
throw (new AssistantApiException(409, "Wait for the assistant to finish before confirming this action."));
|
||||
}
|
||||
|
||||
///////////////////////////////////////////////////////////////////////
|
||||
// claim it first, so a double click cannot run the write twice //
|
||||
///////////////////////////////////////////////////////////////////////
|
||||
|
||||
@@ -160,7 +160,7 @@ public final class TurnStreams
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
private static void writeLive(OutputStream out, TurnEventLog log, int after) throws IOException, InterruptedException
|
||||
static void writeLive(OutputStream out, TurnEventLog log, int after) throws IOException, InterruptedException
|
||||
{
|
||||
int last = after;
|
||||
while(true)
|
||||
@@ -170,7 +170,12 @@ public final class TurnStreams
|
||||
{
|
||||
if(log.isClosed())
|
||||
{
|
||||
return;
|
||||
// The terminal event may have arrived after the empty poll.
|
||||
if(log.size() <= last)
|
||||
{
|
||||
return;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
write(out, ping());
|
||||
continue;
|
||||
|
||||
+1
-1
@@ -80,7 +80,7 @@ public class ConfiguredClassificationSource implements ClassificationSource
|
||||
@Override
|
||||
public DataClassification ofTable(String table)
|
||||
{
|
||||
return (table == null || table.startsWith("assistant_") ? DataClassification.PUBLIC : Classifications.max(null, tables.get(table)));
|
||||
return (table == null || table.startsWith("assistant_") ? DataClassification.PUBLIC : Classifications.max(null, tables.get(table) == null ? AiDataClassifications.ofTable(coreTable(table)) : tables.get(table)));
|
||||
}
|
||||
|
||||
|
||||
|
||||
+115
-23
@@ -44,6 +44,7 @@ import com.kingsrook.qbits.assistant.model.AssistantScheduledTask;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantToolCall;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantTurn;
|
||||
import com.kingsrook.qbits.assistant.persistence.AssistantStore;
|
||||
import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
|
||||
import com.kingsrook.qbits.assistant.scheduled.ScheduledTurn;
|
||||
import com.kingsrook.qbits.assistant.spi.AssistantModelProvider;
|
||||
import com.kingsrook.qbits.assistant.spi.CacheHint;
|
||||
@@ -106,10 +107,7 @@ public class TurnEngine implements Runnable
|
||||
public static final String TIMEOUT_TEXT = "The assistant took too long to answer, so it was stopped. Ask again.";
|
||||
public static final String RESTART_TEXT = "The server restarted while answering. Ask again.";
|
||||
|
||||
private static final int MAX_INPUT_JSON_CHARS = 4000;
|
||||
|
||||
private static final AiRedactor INPUT_REDACTOR = new AiRedactor(List.of());
|
||||
private static final int MAX_RESULT_JSON_CHARS = 64_000;
|
||||
private static final int TEXT_FLUSH_CHARS = 48;
|
||||
private static final long TEXT_FLUSH_MILLIS = 120;
|
||||
|
||||
@@ -155,6 +153,8 @@ public class TurnEngine implements Runnable
|
||||
|
||||
private volatile boolean finished = false;
|
||||
private UiApiCredentials credentials;
|
||||
private volatile TranscriptRedactor transcript;
|
||||
private boolean engineRunning;
|
||||
|
||||
|
||||
|
||||
@@ -200,6 +200,32 @@ public class TurnEngine implements Runnable
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Continue the API's provisional transcript scope on the engine thread.
|
||||
*******************************************************************************/
|
||||
public TurnEngine withTranscript(TranscriptRedactor transcript)
|
||||
{
|
||||
this.transcript = transcript;
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Finalization may run before submission or on a watchdog/shutdown thread.
|
||||
** Bind the same turn scope before any of those paths writes transcript content.
|
||||
*******************************************************************************/
|
||||
private synchronized void attachTranscript()
|
||||
{
|
||||
if(transcript == null)
|
||||
{
|
||||
transcript = new TranscriptRedactor(runtime.getQInstance(), config.getClassificationSource());
|
||||
}
|
||||
transcript.attach();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Run the turn to completion. Never throws.
|
||||
*******************************************************************************/
|
||||
@@ -220,6 +246,15 @@ public class TurnEngine implements Runnable
|
||||
try
|
||||
{
|
||||
QContext.init(runtime.getQInstance(), session);
|
||||
synchronized(this)
|
||||
{
|
||||
if(running.isFinalized())
|
||||
{
|
||||
return;
|
||||
}
|
||||
engineRunning = true;
|
||||
attachTranscript();
|
||||
}
|
||||
startWatchdog();
|
||||
scheduledTask = runtime.getScheduledTurn().task(conversation.getId());
|
||||
if(conversation.getScheduledTaskId() != null && scheduledTask == null)
|
||||
@@ -443,12 +478,22 @@ public class TurnEngine implements Runnable
|
||||
errorCode = stop.errorCode();
|
||||
errorText = stop.message();
|
||||
}
|
||||
finish(status, stopReason, errorCode, errorText);
|
||||
if(interrupted)
|
||||
try
|
||||
{
|
||||
Thread.currentThread().interrupt();
|
||||
finish(status, stopReason, errorCode, errorText);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if(transcript != null)
|
||||
{
|
||||
transcript.close();
|
||||
}
|
||||
if(interrupted)
|
||||
{
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
QContext.clear();
|
||||
}
|
||||
QContext.clear();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -815,6 +860,7 @@ public class TurnEngine implements Runnable
|
||||
for(int i = 0; i < toolUses.size(); i++)
|
||||
{
|
||||
ModelPart use = toolUses.get(i);
|
||||
transcript.input(use.getInput());
|
||||
if(i >= allowed)
|
||||
{
|
||||
limitReached = true;
|
||||
@@ -835,8 +881,14 @@ public class TurnEngine implements Runnable
|
||||
AiToolCall call = new AiToolCall().withId(use.getToolUseId()).withName(use.getToolName()).withInput(use.getInput());
|
||||
futures.add(runtime.getExecutor().submit(() ->
|
||||
{
|
||||
if(!transcript.retainWorker())
|
||||
{
|
||||
return;
|
||||
}
|
||||
transcript.attach();
|
||||
try
|
||||
{
|
||||
QContext.init(runtime.getQInstance(), session);
|
||||
permits.acquire();
|
||||
try
|
||||
{
|
||||
@@ -898,6 +950,12 @@ public class TurnEngine implements Runnable
|
||||
results[index] = new AiToolResult().withToolCallId(call.getId()).withToolName(call.getName()).withOutcome(AiToolOutcome.ERROR)
|
||||
.withSummary("Failed").withModelContent("The tool failed unexpectedly.");
|
||||
}
|
||||
finally
|
||||
{
|
||||
transcript.releaseWorker();
|
||||
TranscriptRedactor.detach();
|
||||
QContext.clear();
|
||||
}
|
||||
}));
|
||||
}
|
||||
persistPendingEvents();
|
||||
@@ -995,6 +1053,7 @@ public class TurnEngine implements Runnable
|
||||
|
||||
try
|
||||
{
|
||||
String inputJson = redactInput(PresentResultsTool.NAME.equals(use.getToolName()) ? (presentation ? result.getRender() : Map.of()) : use.getInput());
|
||||
store.insert(new AssistantToolCall()
|
||||
.withTurnId(turn.getId())
|
||||
.withConversationId(conversation.getId())
|
||||
@@ -1002,7 +1061,7 @@ public class TurnEngine implements Runnable
|
||||
.withToolCallId(use.getToolUseId())
|
||||
.withToolName(use.getToolName())
|
||||
.withClientChannel(scheduledTask == null ? "assistant/" + turn.getId() : "assistant/scheduled/" + scheduledTask.getId())
|
||||
.withInputJson(truncate(redactInput(PresentResultsTool.NAME.equals(use.getToolName()) ? (presentation ? result.getRender() : Map.of()) : use.getInput()), MAX_INPUT_JSON_CHARS))
|
||||
.withInputJson(BulkLoadTools.NAMES.contains(use.getToolName()) ? BulkLoadTools.forLog(inputJson) : inputJson)
|
||||
.withOutcome(outcome)
|
||||
.withPermissionCheck(result.getPermissionCheck())
|
||||
.withTableName(result.getTableName())
|
||||
@@ -1011,7 +1070,7 @@ public class TurnEngine implements Runnable
|
||||
.withRowCount(result.getRowCount())
|
||||
.withDurationMs(result.getDurationMs())
|
||||
.withResultSummary(summary)
|
||||
.withResultJson(truncate(BulkLoadTools.NAMES.contains(use.getToolName()) ? BulkLoadTools.forLog(result.getModelContent()) : result.getModelContent(), MAX_RESULT_JSON_CHARS))
|
||||
.withResultJson(BulkLoadTools.NAMES.contains(use.getToolName()) ? BulkLoadTools.forLog(result.getModelContent()) : result.getModelContent())
|
||||
.withRedactionsJson(result.getRedactions() == null || result.getRedactions().isEmpty() ? null : JsonUtils.toJson(result.getRedactions()))
|
||||
.withUiApiOperation(extras == null ? ("search_knowledge".equals(use.getToolName()) ? "GET /table/{tableName}/{primaryKey}" : null) : extras.getOperation())
|
||||
.withHttpStatus(extras == null ? null : extras.getHttpStatus())
|
||||
@@ -1117,8 +1176,10 @@ public class TurnEngine implements Runnable
|
||||
{
|
||||
return;
|
||||
}
|
||||
TranscriptRedactor previousTranscript = TranscriptRedactor.current();
|
||||
try
|
||||
{
|
||||
attachTranscript();
|
||||
AssistantTurn persistedTurn = null;
|
||||
try
|
||||
{
|
||||
@@ -1164,6 +1225,7 @@ public class TurnEngine implements Runnable
|
||||
if(!alreadyTerminal)
|
||||
{
|
||||
safely(this::persistPendingEvents);
|
||||
safely(() -> transcript.flush(store));
|
||||
// Accounting failures cannot skip the terminal row or local cleanup.
|
||||
safely(() ->
|
||||
{
|
||||
@@ -1171,7 +1233,7 @@ public class TurnEngine implements Runnable
|
||||
.withValue("id", turn.getId())
|
||||
.withValue("status", finalStatus)
|
||||
.withValue("phase", "done")
|
||||
.withValue("stopReason", truncate(finalStopReason, 64))
|
||||
.withValue("stopReason", finalStopReason)
|
||||
.withValue("errorCode", finalErrorCode)
|
||||
.withValue("cancelReason", running.getCancelReason())
|
||||
.withValue("model", model)
|
||||
@@ -1244,6 +1306,20 @@ public class TurnEngine implements Runnable
|
||||
{
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1368,18 +1444,44 @@ public class TurnEngine implements Runnable
|
||||
/*******************************************************************************
|
||||
** Persist queued events.
|
||||
*******************************************************************************/
|
||||
private void persistPendingEvents()
|
||||
private synchronized void persistPendingEvents()
|
||||
{
|
||||
if(!running.isFinalized() || finished)
|
||||
{
|
||||
return;
|
||||
}
|
||||
List<TurnEvent> events = new ArrayList<>();
|
||||
TurnEvent event;
|
||||
while((event = toPersist.poll()) != null)
|
||||
{
|
||||
events.add(event);
|
||||
}
|
||||
StringBuilder text = new StringBuilder();
|
||||
for(TurnEvent queued : events)
|
||||
{
|
||||
if(TurnEvent.TEXT_DELTA.equals(queued.getType()))
|
||||
{
|
||||
text.append(queued.getData().get("text"));
|
||||
}
|
||||
}
|
||||
Boolean firstText = true;
|
||||
for(TurnEvent queued : events)
|
||||
{
|
||||
try
|
||||
{
|
||||
store.insertEvent(turn.getId(), userId, event.getSeq(), event.getType(), event.toJson());
|
||||
String json = queued.toJson();
|
||||
if(TurnEvent.TEXT_DELTA.equals(queued.getType()))
|
||||
{
|
||||
Map<String, Object> data = new LinkedHashMap<>(queued.getData());
|
||||
data.put("text", firstText ? text.toString() : "");
|
||||
firstText = false;
|
||||
json = JsonUtils.toJson(data);
|
||||
}
|
||||
store.insertEvent(turn.getId(), userId, queued.getSeq(), queued.getType(), json);
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
LOG.warn("Could not persist assistant turn event", hipaaSafe(e), logPair("turnId", turn.getId()), logPair("seq", event.getSeq()));
|
||||
LOG.warn("Could not persist assistant turn event", hipaaSafe(e), logPair("turnId", turn.getId()), logPair("seq", queued.getSeq()));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1426,16 +1528,6 @@ public class TurnEngine implements Runnable
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
private static String truncate(String value, int max)
|
||||
{
|
||||
return (value == null || value.length() <= max ? value : value.substring(0, max));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
|
||||
@@ -17,6 +17,7 @@
|
||||
package com.kingsrook.qbits.assistant.engine;
|
||||
|
||||
|
||||
import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantUsage;
|
||||
import com.kingsrook.qbits.assistant.persistence.AssistantStore;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelPricing;
|
||||
@@ -145,6 +146,10 @@ public final class UsageRecorder
|
||||
*******************************************************************************/
|
||||
private static String truncate(String value, int max)
|
||||
{
|
||||
if(value != null && value.length() > max && TranscriptRedactor.current() != null)
|
||||
{
|
||||
TranscriptRedactor.current().truncated(value, value.substring(0, max));
|
||||
}
|
||||
return (value == null || value.length() <= max ? value : value.substring(0, max));
|
||||
}
|
||||
|
||||
|
||||
+10
-5
@@ -17,9 +17,14 @@
|
||||
package com.kingsrook.qbits.assistant.jobs;
|
||||
|
||||
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** STAND-IN(core=QRun-IO/qqq#1210, removeBy=RC.9): core job tracking
|
||||
** (jobs_job tracking rows) will implement this contract; hosts may wire their own.
|
||||
** Publishes assistant answers as jobs. The default, JobTrackingRunTracker,
|
||||
** uses QQQ core job tracking (the Jobs tray); NOOP turns it off, and hosts
|
||||
** may wire their own. Every call is isolated from the turn: a failure here
|
||||
** never affects the answer.
|
||||
*******************************************************************************/
|
||||
public interface AssistantRunTracker
|
||||
{
|
||||
@@ -54,7 +59,7 @@ public interface AssistantRunTracker
|
||||
/*******************************************************************************
|
||||
** Publish a run and return its tracking id.
|
||||
*******************************************************************************/
|
||||
String start(String userId, Integer turnId, Integer conversationId, String label);
|
||||
String start(String userId, Integer turnId, Integer conversationId, String label) throws QException;
|
||||
|
||||
/*******************************************************************************
|
||||
** Publish a changed phase.
|
||||
@@ -64,12 +69,12 @@ public interface AssistantRunTracker
|
||||
/*******************************************************************************
|
||||
** Publish the terminal outcome.
|
||||
*******************************************************************************/
|
||||
void finish(String trackingId, String status, String message);
|
||||
void finish(String trackingId, String status, String message) throws QException;
|
||||
|
||||
/*******************************************************************************
|
||||
** Wire the host's Stop action to the owner's cancellation path.
|
||||
*******************************************************************************/
|
||||
default void onCancel(String trackingId, Runnable cancel)
|
||||
default void onCancel(String trackingId, Runnable cancel) throws QException
|
||||
{
|
||||
}
|
||||
}
|
||||
+170
@@ -0,0 +1,170 @@
|
||||
/*
|
||||
* Copyright 2026 Kingsrook, LLC
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.assistant.jobs;
|
||||
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantConversation;
|
||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.jobs.model.JobStatus;
|
||||
import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
|
||||
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
|
||||
import com.kingsrook.qqq.backend.core.jobs.progress.JobTracking;
|
||||
import com.kingsrook.qqq.backend.core.jobs.progress.TrackedJobRequest;
|
||||
import com.kingsrook.qqq.backend.core.jobs.runtime.DurableJobsRuntime;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Publishes each assistant answer to the Jobs tray as a QQQ tracking job
|
||||
** (core JobTracking, qqq#1210): processName "assistant.turn", phases
|
||||
** thinking / tool / writing, the tool's label as the message, and the tray's
|
||||
** Stop wired to the turn's own Stop. Never prompt or answer text. When the
|
||||
** durable jobs runtime is not running on this pod, answers are simply not
|
||||
** tracked.
|
||||
*******************************************************************************/
|
||||
public class JobTrackingRunTracker implements AssistantRunTracker
|
||||
{
|
||||
public static final String PROCESS_NAME = "assistant.turn";
|
||||
public static final String LABEL = "Assistant answer";
|
||||
|
||||
private final Map<String, JobProgress> reporters = new ConcurrentHashMap<>();
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Start a tracking job linked to the conversation, or null when the jobs
|
||||
** runtime is not running here.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public String start(String userId, Integer turnId, Integer conversationId, String label) throws QException
|
||||
{
|
||||
QInstance qInstance = QContext.getQInstance();
|
||||
if(qInstance == null || DurableJobsRuntime.get(qInstance) == null)
|
||||
{
|
||||
return (null);
|
||||
}
|
||||
|
||||
Integer jobId = JobTracking.start(new TrackedJobRequest()
|
||||
.withUserId(userId)
|
||||
.withLabel(LABEL)
|
||||
.withProcessName(PROCESS_NAME)
|
||||
.withTableName(AssistantConversation.TABLE_NAME)
|
||||
.withRecordId(conversationId));
|
||||
String trackingId = String.valueOf(jobId);
|
||||
reporters.put(trackingId, JobTracking.progress(jobId).phases(
|
||||
JobPhase.of("thinking", "Thinking", 1),
|
||||
JobPhase.of("tool", "Using a tool", 1),
|
||||
JobPhase.of("writing", "Writing", 1)));
|
||||
return (trackingId);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The current phase, with the tool's label as the message.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public void progress(String trackingId, String phase, String message)
|
||||
{
|
||||
JobProgress reporter = trackingId == null ? null : reporters.get(trackingId);
|
||||
if(reporter == null || !("thinking".equals(phase) || "tool".equals(phase) || "writing".equals(phase)))
|
||||
{
|
||||
return;
|
||||
}
|
||||
reporter.phase(phase);
|
||||
if("tool".equals(phase) && message != null)
|
||||
{
|
||||
reporter.message(message);
|
||||
reporter.flush();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** COMPLETE -> COMPLETED, CANCELLED -> CANCELLED, anything else -> FAILED.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public void finish(String trackingId, String status, String message) throws QException
|
||||
{
|
||||
if(trackingId == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
JobProgress reporter = reporters.remove(trackingId);
|
||||
if(reporter != null)
|
||||
{
|
||||
reporter.flush();
|
||||
}
|
||||
JobTracking.finish(Integer.valueOf(trackingId), jobStatus(status), finishMessage(status));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The tray's Stop runs on this pod and stops the turn; the turn's own
|
||||
** finaliser then finishes the job as CANCELLED.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public void onCancel(String trackingId, Runnable cancel) throws QException
|
||||
{
|
||||
if(trackingId != null)
|
||||
{
|
||||
JobTracking.onCancel(Integer.valueOf(trackingId), cancel);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The jobs status for an assistant turn status.
|
||||
*******************************************************************************/
|
||||
static JobStatus jobStatus(String status)
|
||||
{
|
||||
if("COMPLETE".equals(status))
|
||||
{
|
||||
return (JobStatus.COMPLETED);
|
||||
}
|
||||
if("CANCELLED".equals(status))
|
||||
{
|
||||
return (JobStatus.CANCELLED);
|
||||
}
|
||||
return (JobStatus.FAILED);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A fixed message (never the answer, the prompt or a provider error).
|
||||
*******************************************************************************/
|
||||
static String finishMessage(String status)
|
||||
{
|
||||
if("COMPLETE".equals(status))
|
||||
{
|
||||
return ("Answered.");
|
||||
}
|
||||
if("CANCELLED".equals(status))
|
||||
{
|
||||
return ("Stopped.");
|
||||
}
|
||||
return ("The answer did not finish.");
|
||||
}
|
||||
|
||||
}
|
||||
-126
@@ -1,126 +0,0 @@
|
||||
/*
|
||||
* Copyright 2026 Kingsrook, LLC
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.assistant.metadata;
|
||||
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.help.QHelpContent;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Applies help/qbit-assistant.yaml (the QQQ help-bundle format) to the
|
||||
** Assistant app and its assistant_* tables: what each table is for and how
|
||||
** its rows get created, shown on the table screen and when it is empty.
|
||||
** STAND-IN(core=#1198): replace with HelpBundles.register once the pinned
|
||||
** QQQ release includes it; the bundle file stays the same.
|
||||
*******************************************************************************/
|
||||
public final class AssistantHelp
|
||||
{
|
||||
public static final String RESOURCE = "/help/qbit-assistant.yaml";
|
||||
public static final String SLOT = "help";
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Utility class.
|
||||
*******************************************************************************/
|
||||
private AssistantHelp()
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The bundle's entries by key, validated (fails startup if malformed).
|
||||
*******************************************************************************/
|
||||
public static Map<String, JsonNode> entries() throws QException
|
||||
{
|
||||
try(InputStream stream = AssistantHelp.class.getResourceAsStream(RESOURCE))
|
||||
{
|
||||
if(stream == null)
|
||||
{
|
||||
throw (new IOException("missing resource"));
|
||||
}
|
||||
JsonNode bundle = new ObjectMapper(new YAMLFactory()).readTree(stream);
|
||||
if(!"qbit-assistant".equals(bundle.path("bundle").asText()) || !"QBIT".equals(bundle.path("layer").asText()))
|
||||
{
|
||||
throw (new IOException("invalid bundle header"));
|
||||
}
|
||||
Map<String, JsonNode> entries = new LinkedHashMap<>();
|
||||
for(JsonNode entry : bundle.path("entries"))
|
||||
{
|
||||
String key = entry.path("key").asText();
|
||||
String summary = entry.path("summary").asText();
|
||||
if(key.isBlank() || summary.isBlank() || summary.length() > 120 || entry.path("content").asText().isBlank() || entries.put(key, entry) != null)
|
||||
{
|
||||
throw (new IOException("invalid or duplicate entry: " + key));
|
||||
}
|
||||
}
|
||||
return (entries);
|
||||
}
|
||||
catch(IOException | RuntimeException e)
|
||||
{
|
||||
throw (new QException("Could not load " + RESOURCE, e));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Apply the help to the app and every assistant_* table in the instance.
|
||||
** A host's own help on a table is kept (the bundle only fills the slot).
|
||||
*******************************************************************************/
|
||||
public static void apply(QInstance qInstance) throws QException
|
||||
{
|
||||
Map<String, JsonNode> entries = entries();
|
||||
for(QTableMetaData table : qInstance.getTables().values())
|
||||
{
|
||||
JsonNode entry = entries.get("table:" + table.getName() + ";slot:" + SLOT);
|
||||
if(entry != null && (table.getHelpContent() == null || !table.getHelpContent().containsKey(SLOT)))
|
||||
{
|
||||
table.withHelpContent(SLOT, content(entry));
|
||||
}
|
||||
}
|
||||
QAppMetaData app = qInstance.getApp(AssistantMetaDataBuilder.APP_NAME);
|
||||
JsonNode appEntry = entries.get("app:" + AssistantMetaDataBuilder.APP_NAME + ";slot:" + SLOT);
|
||||
if(app != null && appEntry != null && (app.getHelpContent() == null || !app.getHelpContent().containsKey(SLOT)))
|
||||
{
|
||||
app.withHelpContent(SLOT, content(appEntry));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Summary as the first paragraph, then the content, as markdown.
|
||||
*******************************************************************************/
|
||||
static QHelpContent content(JsonNode entry)
|
||||
{
|
||||
return (new QHelpContent().withContentAsMarkdown(entry.path("summary").asText().strip() + "\n\n" + entry.path("content").asText().strip()));
|
||||
}
|
||||
|
||||
}
|
||||
+1
-1
@@ -47,7 +47,7 @@ public class AssistantBudgetOverride extends QRecordEntity
|
||||
@QField()
|
||||
private Instant unlimitedUntil;
|
||||
|
||||
@QField(isRequired = true, maxLength = 1000)
|
||||
@QField(isRequired = true)
|
||||
private String reason;
|
||||
|
||||
@QField(isEditable = false, maxLength = 200)
|
||||
|
||||
+1
-1
@@ -66,7 +66,7 @@ public class AssistantDataAudit extends QRecordEntity
|
||||
@QField(isEditable = false, maxLength = 16)
|
||||
private String decision;
|
||||
|
||||
@QField(isEditable = false, maxLength = 1000)
|
||||
@QField(isEditable = false)
|
||||
private String reason;
|
||||
|
||||
@QField(isEditable = false)
|
||||
|
||||
+1
-2
@@ -22,7 +22,6 @@ import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QField;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecordEntity;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.fields.ValueTooLongBehavior;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
@@ -53,7 +52,7 @@ public class AssistantFeedback extends QRecordEntity
|
||||
@QField(isEditable = false)
|
||||
private Integer rating;
|
||||
|
||||
@QField(isEditable = false, maxLength = 64, valueTooLongBehavior = ValueTooLongBehavior.TRUNCATE_ELLIPSIS)
|
||||
@QField(isEditable = false)
|
||||
private String reason;
|
||||
|
||||
@QField(isEditable = false)
|
||||
|
||||
+3
-4
@@ -22,7 +22,6 @@ import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QField;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecordEntity;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.fields.ValueTooLongBehavior;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
@@ -49,7 +48,7 @@ public class AssistantPendingAction extends QRecordEntity
|
||||
@QField(isEditable = false, maxLength = 255)
|
||||
private String userId;
|
||||
|
||||
@QField(isEditable = false, maxLength = 128)
|
||||
@QField(isEditable = false)
|
||||
private String toolCallId;
|
||||
|
||||
@QField(isEditable = false, maxLength = 64)
|
||||
@@ -67,7 +66,7 @@ public class AssistantPendingAction extends QRecordEntity
|
||||
@QField(isEditable = false)
|
||||
private String paramsJson;
|
||||
|
||||
@QField(isEditable = false, maxLength = 500, valueTooLongBehavior = ValueTooLongBehavior.TRUNCATE_ELLIPSIS)
|
||||
@QField(isEditable = false)
|
||||
private String summary;
|
||||
|
||||
@QField(isEditable = false, maxLength = 16)
|
||||
@@ -76,7 +75,7 @@ public class AssistantPendingAction extends QRecordEntity
|
||||
@QField(isEditable = false)
|
||||
private Integer httpStatus;
|
||||
|
||||
@QField(isEditable = false, maxLength = 500, valueTooLongBehavior = ValueTooLongBehavior.TRUNCATE_ELLIPSIS)
|
||||
@QField(isEditable = false)
|
||||
private String resultSummary;
|
||||
|
||||
@QField(isEditable = false)
|
||||
|
||||
+2
-2
@@ -62,13 +62,13 @@ public class AssistantScheduledRun extends QRecordEntity
|
||||
@QField()
|
||||
private Integer turnId;
|
||||
|
||||
@QField(maxLength = 2000)
|
||||
@QField()
|
||||
private String summary;
|
||||
|
||||
@QField()
|
||||
private Long costMicros;
|
||||
|
||||
@QField(maxLength = 500)
|
||||
@QField()
|
||||
private String error;
|
||||
|
||||
@QField(isEditable = false)
|
||||
|
||||
+2
-2
@@ -55,7 +55,7 @@ public class AssistantToolCall extends QRecordEntity
|
||||
@QField(isEditable = false, maxLength = 255)
|
||||
private String userId;
|
||||
|
||||
@QField(isEditable = false, maxLength = 128)
|
||||
@QField(isEditable = false)
|
||||
private String toolCallId;
|
||||
|
||||
@QField(isEditable = false, maxLength = 64)
|
||||
@@ -85,7 +85,7 @@ public class AssistantToolCall extends QRecordEntity
|
||||
@QField(isEditable = false)
|
||||
private Long durationMs;
|
||||
|
||||
@QField(isEditable = false, maxLength = 500, valueTooLongBehavior = ValueTooLongBehavior.TRUNCATE_ELLIPSIS)
|
||||
@QField(isEditable = false)
|
||||
private String resultSummary;
|
||||
|
||||
@QField(isEditable = false)
|
||||
|
||||
@@ -74,7 +74,7 @@ public class AssistantTurn extends QRecordEntity
|
||||
@QField(isEditable = false, maxLength = 16)
|
||||
private String status;
|
||||
|
||||
@QField(isEditable = false, maxLength = 64)
|
||||
@QField(isEditable = false)
|
||||
private String stopReason;
|
||||
|
||||
@QField(isEditable = false, maxLength = 32)
|
||||
|
||||
+1
-1
@@ -84,7 +84,7 @@ public class AssistantUsage extends QRecordEntity
|
||||
@QField(isEditable = false)
|
||||
private Long costMicros;
|
||||
|
||||
@QField(isEditable = false, maxLength = 100)
|
||||
@QField(isEditable = false)
|
||||
private String requestId;
|
||||
|
||||
@QField(isEditable = false)
|
||||
|
||||
+26
-1
@@ -162,7 +162,13 @@ public class AssistantStore
|
||||
{
|
||||
projectBudgetUsage(record);
|
||||
}
|
||||
return ((T) QRecordEntity.fromQRecord(entity.getClass(), cryptRecord(entity.tableName(), record, false)));
|
||||
rememberTranscript(entity.tableName(), entity.toQRecord(), record.getValue("id"));
|
||||
QRecord returned = cryptRecord(entity.tableName(), record, false);
|
||||
if(TranscriptRedactor.current() != null)
|
||||
{
|
||||
TranscriptRedactor.current().restore(returned, entity.toQRecord());
|
||||
}
|
||||
return ((T) QRecordEntity.fromQRecord(entity.getClass(), returned));
|
||||
}
|
||||
|
||||
|
||||
@@ -173,6 +179,7 @@ public class AssistantStore
|
||||
public void update(QRecordEntity entity) throws QException
|
||||
{
|
||||
requireOwnTable(entity.tableName());
|
||||
rememberTranscript(entity.tableName(), entity.toQRecord(), entity.toQRecord().getValue("id"));
|
||||
UpdateOutput output = new UpdateAction().execute(new UpdateInput(entity.tableName()).withRecord(cryptRecord(entity.tableName(), entity.toQRecord(), true)));
|
||||
QRecord record = output.getRecords().get(0);
|
||||
if(CollectionUtils.nullSafeHasContents(record.getErrors()))
|
||||
@@ -189,6 +196,7 @@ public class AssistantStore
|
||||
public void updateFields(String tableName, QRecord record) throws QException
|
||||
{
|
||||
requireOwnTable(tableName);
|
||||
rememberTranscript(tableName, record, record.getValue("id"));
|
||||
UpdateOutput output = new UpdateAction().execute(new UpdateInput(tableName).withRecord(cryptRecord(tableName, record, true)));
|
||||
QRecord result = output.getRecords().get(0);
|
||||
if(CollectionUtils.nullSafeHasContents(result.getErrors()))
|
||||
@@ -899,6 +907,10 @@ public class AssistantStore
|
||||
}
|
||||
QRecord copy = new QRecord();
|
||||
copy.setValues(new LinkedHashMap<>(record.getValues()));
|
||||
if(encrypt && TranscriptRedactor.current() != null)
|
||||
{
|
||||
copy = TranscriptRedactor.current().prepare(copy);
|
||||
}
|
||||
for(String field : ENCRYPTED_FIELDS.getOrDefault(tableName, List.of()))
|
||||
{
|
||||
if(copy.getValue(field) instanceof String text)
|
||||
@@ -911,6 +923,19 @@ public class AssistantStore
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Stage transcript payloads until all classified values of the turn are known.
|
||||
*******************************************************************************/
|
||||
private void rememberTranscript(String table, QRecord record, Serializable id)
|
||||
{
|
||||
if(TranscriptRedactor.current() != null)
|
||||
{
|
||||
TranscriptRedactor.current().saved(table, record, id);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Retention candidates are paged independently of deletion, so retained PHI cannot starve other rows.
|
||||
*******************************************************************************/
|
||||
|
||||
+495
@@ -0,0 +1,495 @@
|
||||
/*
|
||||
* QQQ - Low-code Application Framework for Engineers.
|
||||
* Copyright (C) 2026. Kingsrook, LLC
|
||||
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
|
||||
* contact@kingsrook.com
|
||||
* https://github.com/Kingsrook/
|
||||
*
|
||||
* 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
|
||||
*
|
||||
* https://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.persistence;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.net.URLEncoder;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
import com.kingsrook.qbits.assistant.classification.ClassificationSource;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantPendingAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
|
||||
import com.kingsrook.qqq.backend.core.actions.permissions.fieldaccess.ClassifiedValueTokens;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.LogRedactorInterface;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
|
||||
import com.kingsrook.qqq.backend.core.utils.JsonUtils;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Turn-local storage egress protection. Values and provisional payloads live only
|
||||
** in memory; QQQ actions receive placeholders until the complete turn is known.
|
||||
** The log hook is global while active, covering provider and tool worker threads.
|
||||
*******************************************************************************/
|
||||
public class TranscriptRedactor implements LogRedactorInterface, AutoCloseable
|
||||
{
|
||||
private static final ThreadLocal<TranscriptRedactor> CURRENT = new ThreadLocal<>();
|
||||
private static final Set<String> PAYLOAD_FIELDS = Set.of("partsJson", "inputJson", "resultJson", "redactionsJson", "dataJson", "paramsJson",
|
||||
"summary", "resultSummary", "reason", "title", "contextJson", "errorMessage", "errorText", "error", "requestId", "comment", "providerRequestId", "providerRequestIds", "auditHref", "toolCallId", "stopReason");
|
||||
private final ClassificationSource classifications;
|
||||
private final QInstance instance;
|
||||
private final Map<String, String> replacements = new LinkedHashMap<>();
|
||||
private final Map<List<String>, String> recordTokens = new LinkedHashMap<>();
|
||||
private final Map<List<String>, QRecord> pending = new LinkedHashMap<>();
|
||||
private Boolean flushing = false;
|
||||
private volatile Boolean closed = false;
|
||||
private volatile Boolean released = false;
|
||||
private Integer workers = 0;
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Register only for this turn; close releases both values and the log hook.
|
||||
*******************************************************************************/
|
||||
public TranscriptRedactor(QInstance instance, ClassificationSource classifications)
|
||||
{
|
||||
this.instance = instance;
|
||||
this.classifications = classifications;
|
||||
QLogger.registerRedactor(this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Explicit propagation works for pooled tool executors, without inheritable state.
|
||||
*******************************************************************************/
|
||||
public void attach()
|
||||
{
|
||||
CURRENT.set(this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Remove only the thread binding, retaining the turn's in-memory state.
|
||||
*******************************************************************************/
|
||||
public static void detach()
|
||||
{
|
||||
CURRENT.remove();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Storage and tools share the currently bound turn, if any.
|
||||
*******************************************************************************/
|
||||
public static TranscriptRedactor current()
|
||||
{
|
||||
return (CURRENT.get());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Capture the raw canonical value and the actual displayed/truncated variant.
|
||||
** Tokens remain bound to the original record field, including equal values.
|
||||
*******************************************************************************/
|
||||
public void observe(String tableName, String fieldName, Serializable primaryKey, Object value, Object displayed)
|
||||
{
|
||||
QTableMetaData table = instance.getTable(tableName);
|
||||
if(released || value == null || table == null || table.getField(fieldName) == null
|
||||
|| !classifications.ofField(tableName, fieldName).isAtLeast(DataClassification.PII))
|
||||
{
|
||||
return;
|
||||
}
|
||||
Serializable original = value instanceof Serializable serializable ? serializable : String.valueOf(value);
|
||||
String token = primaryKey == null
|
||||
? ClassifiedValueTokens.forContext(List.of("assistant", tableName, fieldName), original)
|
||||
: ClassifiedValueTokens.forRecordField(table, table.getField(fieldName), primaryKey, original);
|
||||
synchronized(this)
|
||||
{
|
||||
if(released)
|
||||
{
|
||||
return;
|
||||
}
|
||||
recordTokens.put(List.of(tableName, fieldName, String.valueOf(primaryKey), String.valueOf(displayed)), token);
|
||||
remember(ClassifiedValueTokens.canonical(original), token);
|
||||
remember(String.valueOf(value), token);
|
||||
if(displayed != null)
|
||||
{
|
||||
remember(String.valueOf(displayed), token);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A snippet is another spelling of an already observed field, not a new value.
|
||||
*******************************************************************************/
|
||||
public synchronized void variant(String table, String field, String primaryKey, String text)
|
||||
{
|
||||
if(!closed && text != null && table != null && field != null && primaryKey != null)
|
||||
{
|
||||
recordTokens.entrySet().stream().filter(entry -> entry.getKey().subList(0, 3).equals(List.of(table, field, primaryKey)))
|
||||
.findFirst().ifPresent(entry -> remember(text, entry.getValue()));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Also protect JSON-escaped and URL-encoded occurrences in envelopes and links.
|
||||
*******************************************************************************/
|
||||
private void remember(String value, String token)
|
||||
{
|
||||
if(value == null || value.isEmpty())
|
||||
{
|
||||
return;
|
||||
}
|
||||
replacements.putIfAbsent(value, token);
|
||||
String json = JsonUtils.toJson(value);
|
||||
replacements.putIfAbsent(json.substring(1, json.length() - 1), token);
|
||||
replacements.putIfAbsent(URLEncoder.encode(value, StandardCharsets.UTF_8), token);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Existing tool summaries may clip a value before storage. Remember that exact
|
||||
** prefix as another variant while keeping the live/model summary unchanged.
|
||||
*******************************************************************************/
|
||||
public synchronized void truncated(String original, String prefix)
|
||||
{
|
||||
for(var entry : new ArrayList<>(replacements.entrySet()))
|
||||
{
|
||||
int start = original.indexOf(entry.getKey());
|
||||
while(start >= 0)
|
||||
{
|
||||
if(start < prefix.length() && start + entry.getKey().length() > prefix.length())
|
||||
{
|
||||
remember(prefix.substring(start), entry.getValue());
|
||||
}
|
||||
start = original.indexOf(entry.getKey(), start + 1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Criteria and proposed values are classified even when a query returns no rows.
|
||||
*******************************************************************************/
|
||||
public void input(Map<String, Object> input)
|
||||
{
|
||||
if(input != null && input.get("table") instanceof String table)
|
||||
{
|
||||
criteria(table, input.get("filter"));
|
||||
if(input.get("values") instanceof Map<?, ?> values)
|
||||
{
|
||||
values.forEach((field, value) -> observe(table, String.valueOf(field), null, value, value));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Traverse both model and QQQ filter spellings, including nested boolean groups.
|
||||
*******************************************************************************/
|
||||
private void criteria(String table, Object node)
|
||||
{
|
||||
if(node instanceof Map<?, ?> map)
|
||||
{
|
||||
Object field = map.containsKey("field") ? map.get("field") : map.get("fieldName");
|
||||
if(field instanceof String name && map.get("values") instanceof List<?> values)
|
||||
{
|
||||
values.forEach(value -> observe(table, name, null, value, value));
|
||||
}
|
||||
map.values().forEach(value -> criteria(table, value));
|
||||
}
|
||||
else if(node instanceof List<?> list)
|
||||
{
|
||||
list.forEach(value -> criteria(table, value));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Longest exact match wins; a single pass never redacts a generated token again.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public synchronized String redact(String text)
|
||||
{
|
||||
if(text == null || replacements.isEmpty())
|
||||
{
|
||||
return (text);
|
||||
}
|
||||
List<String> values = new ArrayList<>(replacements.keySet());
|
||||
values.sort(Comparator.comparingInt(String::length).reversed());
|
||||
Pattern pattern = Pattern.compile("(?:hmac|shred):v1:[A-Za-z0-9._-]+:[A-Za-z0-9_-]+|changed:v1|" + String.join("|", values.stream().map(Pattern::quote).toList()));
|
||||
Matcher matcher = pattern.matcher(text);
|
||||
return (matcher.replaceAll(match -> Matcher.quoteReplacement(replacements.getOrDefault(match.group(), match.group()))));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Redact JSON leaves without damaging property names or numeric syntax. Enveloped
|
||||
** tool data retains its framing. Invalid/opaque content is still exact-redacted.
|
||||
*******************************************************************************/
|
||||
public String payload(String text)
|
||||
{
|
||||
if(text == null)
|
||||
{
|
||||
return (null);
|
||||
}
|
||||
int start = text.indexOf('{');
|
||||
int end = text.lastIndexOf('}');
|
||||
try
|
||||
{
|
||||
if(text.stripLeading().startsWith("["))
|
||||
{
|
||||
return (JsonUtils.toJson(tree(JsonUtils.toObject(text, Object.class), null, null)));
|
||||
}
|
||||
if(start >= 0 && end >= start)
|
||||
{
|
||||
Object json = JsonUtils.toObject(text.substring(start, end + 1), Object.class);
|
||||
return (redact(text.substring(0, start)) + JsonUtils.toJson(tree(json, null, null)) + redact(text.substring(end + 1)));
|
||||
}
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
return (redact(text));
|
||||
}
|
||||
return (redact(text));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Preserve record bindings in structured query/get results and render copies.
|
||||
*******************************************************************************/
|
||||
private synchronized Object tree(Object node, String table, String primaryKey)
|
||||
{
|
||||
if(node instanceof Map<?, ?> map)
|
||||
{
|
||||
Object namedTable = map.containsKey("table") ? map.get("table") : map.get("tableName");
|
||||
String resolvedTable = namedTable instanceof String name ? name : table;
|
||||
Object id = map.containsKey("id") ? map.get("id") : map.get("recordId");
|
||||
String resolvedKey = id == null ? primaryKey : String.valueOf(id);
|
||||
Map<String, Object> copy = new LinkedHashMap<>();
|
||||
map.forEach((key, value) ->
|
||||
{
|
||||
String token = recordTokens.get(List.of(String.valueOf(resolvedTable), String.valueOf(key), String.valueOf(resolvedKey), String.valueOf(value)));
|
||||
copy.put(redact(String.valueOf(key)), token == null ? tree(value, resolvedTable, resolvedKey) : token);
|
||||
});
|
||||
return (copy);
|
||||
}
|
||||
if(node instanceof List<?> list)
|
||||
{
|
||||
return (list.stream().map(value -> tree(value, table, primaryKey)).toList());
|
||||
}
|
||||
if(node instanceof String text)
|
||||
{
|
||||
return (redact(text));
|
||||
}
|
||||
if(node instanceof Number && replacements.containsKey(String.valueOf(node)))
|
||||
{
|
||||
return (replacements.get(String.valueOf(node)));
|
||||
}
|
||||
return (node);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Never put provisional content in a backend, its audits, or its SQL logs.
|
||||
*******************************************************************************/
|
||||
public synchronized QRecord prepare(QRecord original)
|
||||
{
|
||||
QRecord copy = new QRecord();
|
||||
copy.setValues(new LinkedHashMap<>(original.getValues()));
|
||||
for(String field : PAYLOAD_FIELDS)
|
||||
{
|
||||
if(copy.getValue(field) instanceof String text)
|
||||
{
|
||||
copy.setValue(field, flushing && !closed ? stored(field, text) : (field.endsWith("Json") ? (field.equals("partsJson") ? "[]" : "{}") : ClassifiedValueTokens.changed()));
|
||||
}
|
||||
}
|
||||
if(!flushing && copy.getValues().containsKey("paramsJson") && !AssistantPendingAction.STATUS_REJECTED.equals(copy.getValueString("status")))
|
||||
{
|
||||
copy.setValue("status", "PREPARING");
|
||||
}
|
||||
return (copy);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Truncate only after redaction, so a storage bound never cuts a raw value.
|
||||
*******************************************************************************/
|
||||
private String stored(String field, String text)
|
||||
{
|
||||
String safe = payload(text);
|
||||
Integer limit = switch(field)
|
||||
{
|
||||
case "inputJson" -> 4000;
|
||||
case "resultJson" -> 64000;
|
||||
case "title" -> 60;
|
||||
default -> null;
|
||||
};
|
||||
if("title".equals(field))
|
||||
{
|
||||
safe = safe.strip().split("\\R", 2)[0].replaceAll("\\s+", " ").strip();
|
||||
}
|
||||
return (limit != null && safe.length() > limit ? safe.substring(0, limit - 1) + "…" : safe);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Return live content to the caller without losing backend-generated defaults.
|
||||
*******************************************************************************/
|
||||
public void restore(QRecord returned, QRecord original)
|
||||
{
|
||||
for(String field : PAYLOAD_FIELDS)
|
||||
{
|
||||
if(original.getValues().containsKey(field))
|
||||
{
|
||||
returned.setValue(field, original.getValue(field));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Remember only content columns, merging later writes without stale bookkeeping.
|
||||
*******************************************************************************/
|
||||
public synchronized void saved(String table, QRecord original, Serializable id)
|
||||
{
|
||||
if(flushing || closed || PAYLOAD_FIELDS.stream().noneMatch(original.getValues()::containsKey))
|
||||
{
|
||||
return;
|
||||
}
|
||||
QRecord content = pending.computeIfAbsent(List.of(table, String.valueOf(id)), key -> new QRecord().withValue("id", id));
|
||||
for(String field : PAYLOAD_FIELDS)
|
||||
{
|
||||
if(original.getValues().containsKey(field))
|
||||
{
|
||||
content.setValue(field, original.getValue(field));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Flush only redacted copies. A crash/failure leaves value-free placeholders.
|
||||
** Redacted proposals cannot later submit token strings to an application field.
|
||||
*******************************************************************************/
|
||||
public void flush(AssistantStore store) throws QException
|
||||
{
|
||||
Map<List<String>, QRecord> writes;
|
||||
Boolean complete;
|
||||
synchronized(this)
|
||||
{
|
||||
complete = workers == 0;
|
||||
flushing = complete;
|
||||
writes = new LinkedHashMap<>(pending);
|
||||
pending.clear();
|
||||
}
|
||||
for(var entry : writes.entrySet())
|
||||
{
|
||||
QRecord record = entry.getValue();
|
||||
String table = entry.getKey().getFirst();
|
||||
if(AssistantPendingAction.TABLE_NAME.equals(table) && record.getValueString("paramsJson") != null)
|
||||
{
|
||||
Boolean changed = !complete || !record.getValueString("paramsJson").equals(payload(record.getValueString("paramsJson")));
|
||||
record.setValue("status", changed ? AssistantPendingAction.STATUS_REJECTED : AssistantPendingAction.STATUS_PENDING);
|
||||
if(changed)
|
||||
{
|
||||
record.setValue("resultSummary", "Classified values were removed. Make this change in the application.");
|
||||
}
|
||||
}
|
||||
store.updateFields(table, record);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A timed-out tool can finish after the engine; retain its log protection.
|
||||
*******************************************************************************/
|
||||
public synchronized Boolean retainWorker()
|
||||
{
|
||||
if(closed)
|
||||
{
|
||||
return (false);
|
||||
}
|
||||
workers++;
|
||||
return (true);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Release a completed worker.
|
||||
*******************************************************************************/
|
||||
public synchronized void releaseWorker()
|
||||
{
|
||||
workers--;
|
||||
releaseIfDone();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Clear classified values only after the engine and all tool workers finish.
|
||||
*******************************************************************************/
|
||||
private void releaseIfDone()
|
||||
{
|
||||
if(closed && workers == 0)
|
||||
{
|
||||
released = true;
|
||||
QLogger.unregisterRedactor(this);
|
||||
replacements.clear();
|
||||
recordTokens.clear();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Always release raw data and unregister, including failed/cancelled turns.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public synchronized void close()
|
||||
{
|
||||
closed = true;
|
||||
pending.clear();
|
||||
releaseIfDone();
|
||||
detach();
|
||||
}
|
||||
}
|
||||
@@ -25,6 +25,7 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import com.kingsrook.qbits.assistant.AssistantRuntime;
|
||||
import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
|
||||
import com.kingsrook.qbits.assistant.engine.PolicyService;
|
||||
import com.kingsrook.qbits.assistant.engine.TurnEngine;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantRagSource;
|
||||
@@ -213,6 +214,11 @@ public class KnowledgeTool
|
||||
title = title.substring(0, Math.min(300, title.length()));
|
||||
String snippet = chunks.get(hit.chunkIndex());
|
||||
snippet = snippet.substring(0, Math.min(1500, snippet.length()));
|
||||
if(TranscriptRedactor.current() != null)
|
||||
{
|
||||
TranscriptRedactor.current().variant(hit.table(), hit.field(), hit.pk(), snippet);
|
||||
TranscriptRedactor.current().variant(hit.table(), source.getTitleFieldName(), hit.pk(), title);
|
||||
}
|
||||
records.add(Map.of("table", hit.table(), "recordId", hit.pk(), "title", title, "snippet", snippet, "score", hit.score(), "href", href));
|
||||
result.getRecordRefs().add(new AiRecordRef().withTableName(hit.table()).withId(hit.pk()).withLabel(title).withHref(href));
|
||||
result.getRedactions().addAll(read.getResult().getRedactions());
|
||||
|
||||
@@ -27,6 +27,7 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.TreeMap;
|
||||
import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
|
||||
import com.kingsrook.qbits.assistant.access.DataAccessMode;
|
||||
import com.kingsrook.qbits.assistant.access.DataAccessService;
|
||||
import com.kingsrook.qbits.assistant.engine.PolicyService;
|
||||
@@ -1445,7 +1446,12 @@ public class UiApiToolbox
|
||||
{
|
||||
text = AiValueMasker.mask(text, field.getType());
|
||||
}
|
||||
rs.put(name, sanitize(text, context, redactions));
|
||||
String visible = sanitize(text, context, redactions);
|
||||
if(TranscriptRedactor.current() != null)
|
||||
{
|
||||
TranscriptRedactor.current().observe(table.getName(), name, (Serializable) raw.get(table.getPrimaryKeyField()), entry.getValue(), visible);
|
||||
}
|
||||
rs.put(name, visible);
|
||||
}
|
||||
return (rs);
|
||||
}
|
||||
@@ -1534,6 +1540,10 @@ public class UiApiToolbox
|
||||
return ("");
|
||||
}
|
||||
String line = text.replaceAll("\\s+", " ").strip();
|
||||
if(line.length() > 300 && TranscriptRedactor.current() != null)
|
||||
{
|
||||
TranscriptRedactor.current().truncated(line, line.substring(0, 299));
|
||||
}
|
||||
return (line.length() <= 300 ? line : line.substring(0, 299) + "…");
|
||||
}
|
||||
|
||||
|
||||
+5
@@ -23,6 +23,7 @@ import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
|
||||
import com.kingsrook.qbits.assistant.access.DataAccessService;
|
||||
import com.kingsrook.qbits.assistant.engine.PolicyService;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelToolDefinition;
|
||||
@@ -823,6 +824,10 @@ public class BulkLoadTools
|
||||
{
|
||||
return ("");
|
||||
}
|
||||
if(text.length() > 300 && TranscriptRedactor.current() != null)
|
||||
{
|
||||
TranscriptRedactor.current().truncated(text, text.substring(0, 297));
|
||||
}
|
||||
return (text.length() > 300 ? text.substring(0, 297) + "..." : text);
|
||||
}
|
||||
|
||||
|
||||
+1
-1
@@ -83,7 +83,7 @@ public class HeaderedUiApiInvoker extends UiApiInvoker
|
||||
{
|
||||
if(!RESERVED_HEADERS.contains(name.toLowerCase(Locale.ROOT)))
|
||||
{
|
||||
builder.header(name, value);
|
||||
builder.setHeader(name, value);
|
||||
}
|
||||
});
|
||||
return (builder.build());
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
+37
-11
@@ -26,11 +26,13 @@ import java.util.Map;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
import java.util.stream.Stream;
|
||||
import com.kingsrook.qbits.assistant.metadata.AssistantHelp;
|
||||
import com.kingsrook.qbits.assistant.metadata.AssistantMetaDataBuilder;
|
||||
import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.help.QHelpContent;
|
||||
import com.kingsrook.qqq.backend.core.actions.help.HelpTargetCatalog;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.help.bundles.HelpBundles;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
|
||||
import com.kingsrook.qqq.backend.core.testing.HelpBundleAssertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
@@ -121,21 +123,45 @@ class AssistantTablesGuardrailTest extends BaseAssistantTest
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Every table and the app have bundle help that says how rows get created.
|
||||
** Every owned target has valid bundle help, even on a host without help storage.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void everyTableHasEmptyStateHelp() throws Exception
|
||||
void everyOwnedTargetHasBundleHelp()
|
||||
{
|
||||
assertThat(AssistantHelp.entries()).allSatisfy((key, entry) -> assertThat(entry.path("summary").asText().length()).as(key).isLessThanOrEqualTo(120));
|
||||
assertThat(qInstance.getTable("helpContent")).isNull();
|
||||
var bundles = HelpBundles.of(qInstance).getBundles();
|
||||
assertThat(bundles).hasSize(1);
|
||||
assertThat(bundles.get(0).getId()).isEqualTo("qbit-assistant");
|
||||
HelpBundleAssertions.forInstance(qInstance)
|
||||
.bundles("help/qbit-assistant.yaml")
|
||||
// Core-generated bulk actions belong to core's bundle, not this QBit.
|
||||
.owning(target -> target.owner() instanceof QProcessMetaData process
|
||||
? qInstance.getTable("assistant_conversation").getSourceQBitName().equals(process.getSourceQBitName())
|
||||
: target.name() != null && target.name().startsWith("assistant"))
|
||||
.assertComplete();
|
||||
// RC.7's gate requires tables, fields, apps and processes; also cover our selectors and widgets.
|
||||
for(var target : HelpTargetCatalog.collect(qInstance))
|
||||
{
|
||||
if(List.of("widget", "possibleValueSource").contains(target.kind()) && target.name().startsWith("assistant"))
|
||||
{
|
||||
assertThat(bundles.get(0).getEntries()).as(target.key())
|
||||
.anyMatch(entry -> entry.getKey().equals(target.key()) && entry.getSummary() != null && !entry.getSummary().isBlank());
|
||||
}
|
||||
}
|
||||
for(String table : assistantTables())
|
||||
{
|
||||
List<QHelpContent> help = qInstance.getTable(table).getHelpContent() == null ? null : qInstance.getTable(table).getHelpContent().get(AssistantHelp.SLOT);
|
||||
assertThat(help).as(table + " table help").isNotEmpty();
|
||||
assertThat(help.get(0).getContent()).as(table + " help says how rows get created").contains("How rows get created");
|
||||
assertThat(bundles.get(0).getEntries())
|
||||
.anySatisfy(entry ->
|
||||
{
|
||||
assertThat(entry.getKey()).isEqualTo("table:" + table + ";slot:help");
|
||||
assertThat(entry.getContent()).isNotBlank();
|
||||
});
|
||||
}
|
||||
assertThat(qInstance.getApp(AssistantMetaDataBuilder.APP_NAME).getHelpContent()).containsKey(AssistantHelp.SLOT);
|
||||
assertThat(AssistantHelp.entries().keySet()).allMatch(key -> !key.startsWith("table:") || qInstance.getTable(key.substring(6, key.indexOf(';'))) != null,
|
||||
"help names only real tables");
|
||||
assertThat(bundles.get(0).getEntries()).anySatisfy(entry ->
|
||||
{
|
||||
assertThat(entry.getKey()).isEqualTo("app:" + AssistantMetaDataBuilder.APP_NAME + ";slot:help");
|
||||
assertThat(entry.getContent()).isNotBlank();
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
/*
|
||||
* 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;
|
||||
|
||||
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.sql.DriverManager;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.regex.Pattern;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Execute the payload column definitions and upgrade statements on H2. Other
|
||||
** schema constraints have their own Postgres guardrail; this checks expansion.
|
||||
*******************************************************************************/
|
||||
class TranscriptSchemaTest
|
||||
{
|
||||
/*******************************************************************************
|
||||
** Both fresh installs and upgraded columns accept expanded token sequences.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void redactedColumnsAcceptExpandedTokens() throws Exception
|
||||
{
|
||||
Map<String, List<String>> columns = Map.of(
|
||||
"assistant_turn", List.of("stop_reason"),
|
||||
"assistant_tool_call", List.of("tool_call_id", "result_summary"),
|
||||
"assistant_feedback", List.of("reason"),
|
||||
"assistant_usage", List.of("request_id"),
|
||||
"assistant_pending_action", List.of("tool_call_id", "summary", "result_summary"),
|
||||
"assistant_data_audit", List.of("reason"),
|
||||
"assistant_budget_override", List.of("reason"),
|
||||
"assistant_scheduled_run", List.of("summary", "error"));
|
||||
String ddl = Files.readString(Path.of("..", "docs", "schema-postgres.sql"));
|
||||
String tokens = "hmac:v1:test:0123456789abcdefghijklmnopqrstuvwxyzABCDEFG".repeat(100);
|
||||
try(var connection = DriverManager.getConnection("jdbc:h2:mem:transcript" + System.nanoTime() + ";MODE=PostgreSQL");
|
||||
var statement = connection.createStatement())
|
||||
{
|
||||
for(var table : columns.entrySet())
|
||||
{
|
||||
var create = Pattern.compile("CREATE TABLE " + table.getKey() + "\\s*\\((.*?)\\n\\);", Pattern.DOTALL).matcher(ddl);
|
||||
assertThat(create.find()).as(table.getKey()).isTrue();
|
||||
for(String column : table.getValue())
|
||||
{
|
||||
var definition = Pattern.compile("(?m)^ " + column + " (TEXT|VARCHAR\\(\\d+\\))").matcher(create.group(1));
|
||||
assertThat(definition.find()).as(column).isTrue();
|
||||
statement.execute("DROP TABLE IF EXISTS " + table.getKey());
|
||||
statement.execute("CREATE TABLE " + table.getKey() + " (" + column + " " + definition.group(1) + ")");
|
||||
try(var insert = connection.prepareStatement("INSERT INTO " + table.getKey() + " VALUES (?)"))
|
||||
{
|
||||
insert.setString(1, tokens);
|
||||
assertThat(insert.executeUpdate()).isEqualTo(1);
|
||||
}
|
||||
String migration = "ALTER TABLE " + table.getKey() + " ALTER COLUMN " + column + " SET DATA TYPE TEXT";
|
||||
if(!"result_summary".equals(column) && !("assistant_pending_action".equals(table.getKey()) && "summary".equals(column)))
|
||||
{
|
||||
assertThat(ddl).contains(migration + ";");
|
||||
statement.execute("DELETE FROM " + table.getKey());
|
||||
statement.execute("ALTER TABLE " + table.getKey() + " ALTER COLUMN " + column + " SET DATA TYPE VARCHAR(64)");
|
||||
statement.execute(migration);
|
||||
statement.execute(migration);
|
||||
try(var insert = connection.prepareStatement("INSERT INTO " + table.getKey() + " VALUES (?)"))
|
||||
{
|
||||
insert.setString(1, tokens);
|
||||
assertThat(insert.executeUpdate()).isEqualTo(1);
|
||||
}
|
||||
}
|
||||
try(var result = statement.executeQuery("SELECT " + column + " FROM " + table.getKey()))
|
||||
{
|
||||
assertThat(result.next()).isTrue();
|
||||
assertThat(result.getString(1)).isEqualTo(tokens);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
/*
|
||||
* 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.api;
|
||||
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import com.kingsrook.qbits.assistant.engine.TurnEvent;
|
||||
import com.kingsrook.qbits.assistant.engine.TurnEventLog;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Terminal events must be drained even when closure races an empty poll.
|
||||
*******************************************************************************/
|
||||
class TurnStreamsTest
|
||||
{
|
||||
/*******************************************************************************
|
||||
** Reproduce the poll/close boundary without relying on thread timing.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void drainsTerminalEventsAfterEmptyPoll() throws Exception
|
||||
{
|
||||
TurnEventLog log = new TurnEventLog()
|
||||
{
|
||||
/*******************************************************************************
|
||||
** Append just after a timed-out poll, before the reader checks closure.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public List<TurnEvent> after(int afterSeq, long timeoutMillis) throws InterruptedException
|
||||
{
|
||||
if(size() == 0)
|
||||
{
|
||||
append(TurnEvent.ERROR, Map.of("code", "TIMEOUT"));
|
||||
append(TurnEvent.MESSAGE_END, Map.of("status", "ERROR"));
|
||||
return (List.of());
|
||||
}
|
||||
return (super.after(afterSeq, timeoutMillis));
|
||||
}
|
||||
};
|
||||
ByteArrayOutputStream output = new ByteArrayOutputStream();
|
||||
TurnStreams.writeLive(output, log, 0);
|
||||
assertThat(output.toString(StandardCharsets.UTF_8)).contains("event: error", "TIMEOUT", "event: message.end");
|
||||
}
|
||||
}
|
||||
+83
@@ -37,6 +37,7 @@ import com.kingsrook.qbits.assistant.model.AssistantConversation;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantMessage;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantTurn;
|
||||
import com.kingsrook.qbits.assistant.persistence.AssistantStore;
|
||||
import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
|
||||
import com.kingsrook.qbits.assistant.provider.fake.ScriptedModelProvider;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelEventListener;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelPart;
|
||||
@@ -46,14 +47,18 @@ import com.kingsrook.qbits.assistant.spi.ModelUsage;
|
||||
import com.kingsrook.qbits.assistant.spi.StreamHandle;
|
||||
import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
|
||||
import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
|
||||
import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
|
||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
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.model.data.QRecordEntity;
|
||||
import com.kingsrook.qqq.backend.core.model.session.QSession;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import static com.kingsrook.qbits.assistant.testing.AssistantTestFixture.EDITOR;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
@@ -64,6 +69,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
*******************************************************************************/
|
||||
class DurableTurnsTest extends BaseAssistantTest
|
||||
{
|
||||
private static final String CLASSIFIED_TEXT = "CLASSIFIED-LIFECYCLE-CANARY";
|
||||
|
||||
private QSession session;
|
||||
private AssistantApi api;
|
||||
private AssistantStore store;
|
||||
@@ -108,6 +115,73 @@ class DurableTurnsTest extends BaseAssistantTest
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Terminal paths on other threads flush the same provisional transcript scope.
|
||||
** Live text stays intact; stored user/message/replay content has only tokens.
|
||||
*******************************************************************************/
|
||||
@ParameterizedTest
|
||||
@ValueSource(strings = { "SUPERSEDED", "TIMEOUT", "SHUTDOWN" })
|
||||
void terminalPathsRedactClassifiedContent(String reason) throws Exception
|
||||
{
|
||||
config.withDataClassification("blog_post", "title", DataClassification.PII);
|
||||
ControlledProvider controlled = controlled(true);
|
||||
controlled.classifiedText = CLASSIFIED_TEXT;
|
||||
if("TIMEOUT".equals(reason))
|
||||
{
|
||||
config.setMaxTurnSeconds(1);
|
||||
}
|
||||
Integer conversation = conversation();
|
||||
RunningTurn run = api.startTurn(conversation, Map.of("text", CLASSIFIED_TEXT));
|
||||
try
|
||||
{
|
||||
controlled.awaitCalls(1);
|
||||
if("SHUTDOWN".equals(reason))
|
||||
{
|
||||
config.getRuntime().shutdown(Duration.ZERO);
|
||||
QContext.init(qInstance, session);
|
||||
}
|
||||
else if("SUPERSEDED".equals(reason))
|
||||
{
|
||||
api.startTurn(conversation, Map.of("text", "next question"));
|
||||
}
|
||||
AssistantTurn row = completed(run);
|
||||
assertThat(row.getCancelReason()).isEqualTo(reason);
|
||||
assertThat(store.getMessage(EDITOR, row.getUserMessageId()).getPartsJson()).contains("changed:v1").doesNotContain(CLASSIFIED_TEXT);
|
||||
assertThat(store.getMessage(EDITOR, row.getAssistantMessageId()).getPartsJson()).contains("changed:v1").doesNotContain(CLASSIFIED_TEXT);
|
||||
assertThat(api.turnEvents(run.getTurnId(), 0).persisted()).anySatisfy(event -> assertThat(event.toJson()).contains("changed:v1"))
|
||||
.allSatisfy(event -> assertThat(event.toJson()).doesNotContain(CLASSIFIED_TEXT));
|
||||
assertThat(run.getLog().after(0, 1)).anySatisfy(event -> assertThat(event.toJson()).contains(CLASSIFIED_TEXT));
|
||||
assertThat(TranscriptRedactor.current()).isNull();
|
||||
assertThat(QLogger.redact(CLASSIFIED_TEXT)).isEqualTo("changed:v1");
|
||||
}
|
||||
finally
|
||||
{
|
||||
controlled.release();
|
||||
}
|
||||
await(() -> CLASSIFIED_TEXT.equals(QLogger.redact(CLASSIFIED_TEXT)));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A rejected submission still persists its terminal notice and releases scope.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void rejectedSubmissionFinalizesTranscript() throws Exception
|
||||
{
|
||||
Integer conversation = conversation();
|
||||
config.getRuntime().getExecutor().shutdown();
|
||||
assertThatThrownBy(() -> api.startTurn(conversation, Map.of("text", "Submission failed")))
|
||||
.isInstanceOf(java.util.concurrent.RejectedExecutionException.class);
|
||||
QContext.init(qInstance, session);
|
||||
assertThat(store.listRunningTurns(EDITOR)).isEmpty();
|
||||
assertThat(store.listMessages(EDITOR, conversation, null, 100)).anySatisfy(message ->
|
||||
assertThat(message.getPartsJson()).contains(TurnEngine.RESTART_TEXT));
|
||||
assertThat(TranscriptRedactor.current()).isNull();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Completion closes the lease, emits progress, and frees the next question.
|
||||
*******************************************************************************/
|
||||
@@ -363,11 +437,14 @@ class DurableTurnsTest extends BaseAssistantTest
|
||||
void orphanRecoveryAndIdempotentStop() throws Exception
|
||||
{
|
||||
AssistantTurn stale = remote(Instant.now().minusSeconds(5));
|
||||
store.updateFields(AssistantMessage.TABLE_NAME, new QRecord().withValue("id", stale.getAssistantMessageId())
|
||||
.withValue("partsJson", new MessageParts().addText("changed:v1").toJson()));
|
||||
AssistantTurn fresh = remote(Instant.now());
|
||||
AssistantTurn deadline = remote(Instant.now());
|
||||
store.updateFields(AssistantTurn.TABLE_NAME, new QRecord().withValue("id", deadline.getId()).withValue("startedAt", Instant.now().minusSeconds(30)));
|
||||
assertThat(active()).extracting(r -> r.get("turnId")).containsExactly(fresh.getId());
|
||||
assertThat(store.getTurn(EDITOR, stale.getId()).getStatus()).isEqualTo("ERROR");
|
||||
assertThat(store.getMessage(EDITOR, stale.getAssistantMessageId()).getPartsJson()).contains("changed:v1", "server restarted");
|
||||
assertThat(store.getTurn(EDITOR, deadline.getId()).getStatus()).isEqualTo("ERROR");
|
||||
assertThat(api.turnEvents(stale.getId(), 0).persisted()).extracting(TurnEvent::getType).containsExactly(TurnEvent.ERROR, TurnEvent.MESSAGE_END);
|
||||
assertThat(api.turnEvents(stale.getId(), 0).persisted()).hasSize(2);
|
||||
@@ -724,6 +801,7 @@ class DurableTurnsTest extends BaseAssistantTest
|
||||
private static class ControlledProvider extends ScriptedModelProvider
|
||||
{
|
||||
private final boolean stubborn;
|
||||
private String classifiedText;
|
||||
private final List<ControlledHandle> handles = new CopyOnWriteArrayList<>();
|
||||
private boolean emitText = true;
|
||||
|
||||
@@ -742,6 +820,11 @@ class DurableTurnsTest extends BaseAssistantTest
|
||||
public StreamHandle stream(ModelRequest request, ModelEventListener listener)
|
||||
{
|
||||
ControlledHandle handle = new ControlledHandle(stubborn);
|
||||
if(classifiedText != null)
|
||||
{
|
||||
TranscriptRedactor.current().observe("blog_post", "title", 1, classifiedText, classifiedText);
|
||||
listener.onTextDelta(classifiedText);
|
||||
}
|
||||
if(emitText)
|
||||
{
|
||||
listener.onTextDelta("A partial answer that has already begun streaming to the user. ");
|
||||
|
||||
+1
-1
@@ -133,7 +133,7 @@ class PhiRoutingTest extends BaseAssistantTest
|
||||
assertThat(outcome.toolCalls()).singleElement().satisfies(call ->
|
||||
{
|
||||
assertThat(call.getOutcome()).isEqualTo("OK");
|
||||
assertThat(call.getResultJson()).contains("PHI_DIAGNOSIS_SENTINEL");
|
||||
assertThat(call.getResultJson()).doesNotContain("PHI_DIAGNOSIS_SENTINEL").contains("changed:v1");
|
||||
});
|
||||
assertThat(provider.getRequests()).hasSize(1);
|
||||
assertThat(audits()).singleElement().satisfies(row ->
|
||||
|
||||
+476
@@ -0,0 +1,476 @@
|
||||
/*
|
||||
* QQQ - Low-code Application Framework for Engineers.
|
||||
* Copyright (C) 2026. Kingsrook, LLC
|
||||
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
|
||||
* contact@kingsrook.com
|
||||
* https://github.com/Kingsrook/
|
||||
*
|
||||
* 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
|
||||
*
|
||||
* https://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.engine;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.math.BigDecimal;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import com.kingsrook.qbits.assistant.AssistantQBitConfig;
|
||||
import com.kingsrook.qbits.assistant.api.AssistantApi;
|
||||
import com.kingsrook.qbits.assistant.classification.ConfiguredClassificationSource;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantDataAccess;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantDataAudit;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantPendingAction;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantProvider;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantScheduledRun;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantToolCall;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantTurn;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantTurnEvent;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantUsage;
|
||||
import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
|
||||
import com.kingsrook.qbits.assistant.spi.AssistantModelProvider;
|
||||
import com.kingsrook.qbits.assistant.spi.AssistantProviderType;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelPart;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelProviderException;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelResponse;
|
||||
import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
|
||||
import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
|
||||
import com.kingsrook.qqq.backend.core.actions.ai.AiFieldMetaData;
|
||||
import com.kingsrook.qqq.backend.core.actions.ai.AiTableMetaData;
|
||||
import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
|
||||
import com.kingsrook.qqq.backend.core.actions.permissions.fieldaccess.ClassifiedValueKeySource;
|
||||
import com.kingsrook.qqq.backend.core.actions.permissions.fieldaccess.ClassifiedValueKeySourceMetaData;
|
||||
import com.kingsrook.qqq.backend.core.actions.permissions.fieldaccess.ClassifiedValueTokens;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
|
||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.audits.AuditLevel;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.audits.QAuditRules;
|
||||
import com.kingsrook.qqq.backend.core.utils.JsonUtils;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.core.LogEvent;
|
||||
import org.apache.logging.log4j.core.Logger;
|
||||
import org.apache.logging.log4j.core.appender.AbstractAppender;
|
||||
import org.apache.logging.log4j.core.config.Property;
|
||||
import org.apache.logging.log4j.core.layout.PatternLayout;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Tag;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Secured-data egress canaries: inspect actual backend rows, core audits and logs,
|
||||
** while independently proving that the eligible model still received the values.
|
||||
*******************************************************************************/
|
||||
@Tag("redteam")
|
||||
class TranscriptRedactionTest extends BaseAssistantTest
|
||||
{
|
||||
private final String phi = "CANARY-PHI-" + UUID.randomUUID();
|
||||
private final String pii = "CANARY-PII-" + UUID.randomUUID();
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Enable the fixture table and a local eligible provider.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
protected void customizeConfig(AssistantQBitConfig config)
|
||||
{
|
||||
config.withDataClassification("patient", DataClassification.PHI).withProviderType(new PatientProviderType());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Real core metadata, including table fallback, drives the redactor and core audits.
|
||||
*******************************************************************************/
|
||||
@BeforeEach
|
||||
void seed() throws Exception
|
||||
{
|
||||
admin();
|
||||
var table = qInstance.getTable("patient");
|
||||
table.withSupplementalMetaData(new AiTableMetaData().withClassification(DataClassification.PHI));
|
||||
table.getField("email").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PII));
|
||||
table.getField("id").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PUBLIC));
|
||||
table.getField("status").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PUBLIC));
|
||||
table.withAuditRules(new QAuditRules().withAuditLevel(AuditLevel.FIELD));
|
||||
config.withClassificationSource(new ConfiguredClassificationSource());
|
||||
assertThat(config.getClassificationSource().ofTable("patient")).isEqualTo(DataClassification.PHI);
|
||||
new UpdateAction().execute(new UpdateInput("patient").withRecord(new QRecord().withValue("id", 1).withValue("email", pii).withValue("diagnosis", phi)));
|
||||
config.getRuntime().getStore().insert(new AssistantProvider().withName("Eligible local provider").withProviderType("patient-test")
|
||||
.withModel("test-model").withLocation("LOCAL").withHipaaEligible(true).withEnabled(true).withIsDefault(true));
|
||||
config.getRuntime().getStore().insert(new AssistantDataAccess().withObjectType("TABLE").withObjectName("patient").withMode("READ_WRITE"));
|
||||
config.getRuntime().getDataAccessService().invalidate();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Scan raw backend storage as an administrator, not just decoded transcript APIs.
|
||||
*******************************************************************************/
|
||||
private void admin()
|
||||
{
|
||||
QContext.init(qInstance, AssistantTestFixture.ownerSession(AssistantTestFixture.ADMIN, config));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Every five-character window is sufficient to detect any leaked substring > 4.
|
||||
*******************************************************************************/
|
||||
private void assertNoCanaries(String text)
|
||||
{
|
||||
for(String canary : List.of(phi, pii))
|
||||
{
|
||||
for(int i = 0; i <= canary.length() - 5; i++)
|
||||
{
|
||||
assertThat(text).doesNotContain(canary.substring(i, i + 5));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Query every registered assistant table and all core audit tables, without a row cap.
|
||||
*******************************************************************************/
|
||||
private void scanStorage() throws Exception
|
||||
{
|
||||
admin();
|
||||
for(String table : qInstance.getTables().keySet())
|
||||
{
|
||||
if(table.startsWith("assistant_") || table.toLowerCase().startsWith("audit"))
|
||||
{
|
||||
var rows = new QueryAction().execute(new QueryInput(table).withFilter(new QQueryFilter())).getRecords();
|
||||
assertNoCanaries(JsonUtils.toJson(rows));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Query, get, input filters, presentation, split deltas, proposals and errors.
|
||||
** Log capture includes exception detail and a provider thread without QContext.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void canariesNeverReachStorageOrLogs() throws Exception
|
||||
{
|
||||
List<String> logs = new CopyOnWriteArrayList<>();
|
||||
AbstractAppender appender = new AbstractAppender("transcript-canary", null, PatternLayout.createDefaultLayout(), false, Property.EMPTY_ARRAY)
|
||||
{
|
||||
/*******************************************************************************
|
||||
** Retain the emitted log representation, including any exception.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public void append(LogEvent event)
|
||||
{
|
||||
logs.add(event.getMessage().getFormattedMessage() + (event.getThrown() == null ? "" : event.getThrown().toString()));
|
||||
}
|
||||
};
|
||||
appender.start();
|
||||
Logger logger = (Logger) LogManager.getRootLogger();
|
||||
logger.addAppender(appender);
|
||||
try
|
||||
{
|
||||
provider.thenToolUse("query_records", Map.of("table", "patient"))
|
||||
.thenToolUse("get_record", Map.of("table", "patient", "id", "1"))
|
||||
.thenRespond(request ->
|
||||
{
|
||||
assertThat(JsonUtils.toJson(request)).contains(phi, pii);
|
||||
QLogger.getLogger(TranscriptRedactionTest.class).warn("Provider echoed " + phi + " " + pii, new IllegalStateException(phi + " " + pii));
|
||||
return (new ModelResponse().withParts(List.of(ModelPart.toolUse("present", "present_results", Map.of("summary", phi + " " + pii))))
|
||||
.withStopReason(ModelResponse.STOP_TOOL_USE));
|
||||
})
|
||||
.thenToolUse("query_records", Map.of("table", "patient", "filter", filter("AND", "email", "EQUALS", List.of(pii))))
|
||||
.thenToolUse("update_record", Map.of("table", "patient", "id", "1", "values", Map.of("diagnosis", phi, "email", pii)))
|
||||
.thenText("x".repeat(43) + phi + " " + pii);
|
||||
var outcome = runner.run(AssistantTestFixture.ADMIN, phi + " " + pii);
|
||||
assertThat(outcome.errorCode()).isNull();
|
||||
assertThat(outcome.text()).contains(phi, pii);
|
||||
assertThat(outcome.toolCalls()).hasSize(5).allSatisfy(call -> assertThat(call.getOutcome()).isEqualTo("OK"));
|
||||
admin();
|
||||
var store = config.getRuntime().getStore();
|
||||
var proposals = store.query(AssistantPendingAction.TABLE_NAME, new QQueryFilter(), AssistantPendingAction.class);
|
||||
assertThat(proposals).singleElement().satisfies(action ->
|
||||
{
|
||||
assertThat(action.getParamsJson()).contains("changed:v1");
|
||||
assertThat(action.getStatus()).isEqualTo(AssistantPendingAction.STATUS_REJECTED);
|
||||
});
|
||||
assertThatThrownBy(() -> new AssistantApi(config.getRuntime(), QContext.getQSession()).confirmAction(proposals.getFirst().getId()))
|
||||
.hasMessageContaining("rejected");
|
||||
var events = store.query(AssistantTurnEvent.TABLE_NAME, new QQueryFilter(), AssistantTurnEvent.class);
|
||||
assertThat(events).isNotEmpty();
|
||||
assertThat(events.stream().map(AssistantTurnEvent::getDataJson).reduce("", String::concat)).contains("changed:v1");
|
||||
scanStorage();
|
||||
|
||||
provider.thenToolUse("get_record", Map.of("table", "patient", "id", "1"))
|
||||
.thenError(ModelProviderException.Kind.INVALID_REQUEST, phi + " " + pii);
|
||||
assertThat(runner.run(AssistantTestFixture.ADMIN, "Read patients again").errorCode()).isEqualTo("PROVIDER");
|
||||
scanStorage();
|
||||
assertThat(logs).isNotEmpty();
|
||||
assertNoCanaries(String.join("\n", logs));
|
||||
assertThat(QLogger.redact(phi)).as("turn hook released").isEqualTo(phi);
|
||||
}
|
||||
finally
|
||||
{
|
||||
logger.removeAppender(appender);
|
||||
appender.stop();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Content written before discovery never reaches storage, even transiently.
|
||||
** Includes the audit reason, scheduled summary, bulk log and redactions payloads.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void provisionalPayloadsAndNoResultCriteria() throws Exception
|
||||
{
|
||||
admin();
|
||||
var store = config.getRuntime().getStore();
|
||||
try(TranscriptRedactor redactor = new TranscriptRedactor(qInstance, config.getClassificationSource()))
|
||||
{
|
||||
redactor.attach();
|
||||
var audit = store.insert(new AssistantDataAudit().withUserId(AssistantTestFixture.ADMIN).withReason(phi + " " + pii));
|
||||
store.insert(new AssistantScheduledRun().withUserId(AssistantTestFixture.ADMIN).withTaskId(1).withScheduledFor(Instant.now()).withSummary(phi));
|
||||
store.insert(new AssistantToolCall().withUserId(AssistantTestFixture.ADMIN).withToolName("bulk_load_preview")
|
||||
.withInputJson(JsonUtils.toJson(Map.of("email", pii))).withResultJson(phi).withRedactionsJson(JsonUtils.toJson(List.of(phi, pii))));
|
||||
scanStorage();
|
||||
redactor.input(Map.of("table", "patient", "filter", Map.of("subFilters", List.of(filter("AND", "email", "EQUALS", List.of(pii))))));
|
||||
redactor.observe("patient", "diagnosis", 1, phi, phi);
|
||||
store.updateFields(AssistantDataAudit.TABLE_NAME, new QRecord().withValue("id", audit.getId()).withValue("reason", pii + " " + phi));
|
||||
redactor.flush(store);
|
||||
scanStorage();
|
||||
}
|
||||
provider.thenToolUse("query_records", Map.of("table", "patient", "filter", filter("AND", "email", "EQUALS", List.of(pii + "-absent"))))
|
||||
.thenText(pii + "-absent");
|
||||
runner.run(AssistantTestFixture.ADMIN, "Find missing patient");
|
||||
scanStorage();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Metadata must not reject or clip tokens after the SQL columns are widened.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void expandedTokensSurviveMetadataValidation() throws Exception
|
||||
{
|
||||
admin();
|
||||
var store = config.getRuntime().getStore();
|
||||
String raw = phi.repeat(250);
|
||||
try(TranscriptRedactor redactor = new TranscriptRedactor(qInstance, config.getClassificationSource()))
|
||||
{
|
||||
redactor.attach();
|
||||
redactor.observe("patient", "diagnosis", 1, phi, phi);
|
||||
var audit = store.insert(new AssistantDataAudit().withUserId(AssistantTestFixture.ADMIN).withReason(raw));
|
||||
var turn = store.insert(new AssistantTurn().withUserId(AssistantTestFixture.ADMIN).withStatus("RUNNING").withStopReason(raw));
|
||||
var call = store.insert(new AssistantToolCall().withUserId(AssistantTestFixture.ADMIN).withToolCallId(raw).withResultSummary(raw));
|
||||
var usage = store.insert(new AssistantUsage().withUserId(AssistantTestFixture.ADMIN).withRequestId(raw));
|
||||
redactor.flush(store);
|
||||
String expected = "changed:v1".repeat(250);
|
||||
assertThat(store.getTurn(AssistantTestFixture.ADMIN, turn.getId()).getStopReason()).isEqualTo(expected);
|
||||
for(var check : List.of(Map.entry(audit, "reason"), Map.entry(call, "toolCallId"), Map.entry(call, "resultSummary"), Map.entry(usage, "requestId")))
|
||||
{
|
||||
var row = new QueryAction().execute(new QueryInput(check.getKey().tableName())
|
||||
.withFilter(new QQueryFilter().withCriteria("id", com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator.EQUALS,
|
||||
check.getKey().toQRecord().getValue("id")))).getRecords().getFirst();
|
||||
assertThat(row.getValueString(check.getValue())).isEqualTo(expected);
|
||||
}
|
||||
}
|
||||
scanStorage();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A late tool retains the log hook; incomplete/cancelled turns keep placeholders
|
||||
** and cannot turn a provisional pending payload into an executable empty write.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void lateWorkerKeepsProtectionAndFailsClosed() throws Exception
|
||||
{
|
||||
admin();
|
||||
var store = config.getRuntime().getStore();
|
||||
TranscriptRedactor redactor = new TranscriptRedactor(qInstance, config.getClassificationSource());
|
||||
try
|
||||
{
|
||||
redactor.attach();
|
||||
redactor.observe("patient", "diagnosis", 1, phi, phi);
|
||||
assertThat(redactor.retainWorker()).isTrue();
|
||||
var action = store.insert(new AssistantPendingAction().withUserId(AssistantTestFixture.ADMIN)
|
||||
.withStatus(AssistantPendingAction.STATUS_PENDING).withParamsJson(JsonUtils.toJson(Map.of("diagnosis", phi))));
|
||||
assertThat(store.getPendingAction(AssistantTestFixture.ADMIN, action.getId()).getStatus()).isEqualTo("PREPARING");
|
||||
redactor.flush(store);
|
||||
assertThat(store.getPendingAction(AssistantTestFixture.ADMIN, action.getId()).getStatus()).isEqualTo(AssistantPendingAction.STATUS_REJECTED);
|
||||
redactor.close();
|
||||
assertThat(redactor.retainWorker()).isFalse();
|
||||
assertThat(QLogger.redact(phi)).isEqualTo("changed:v1");
|
||||
redactor.releaseWorker();
|
||||
assertThat(QLogger.redact(phi)).isEqualTo(phi);
|
||||
scanStorage();
|
||||
}
|
||||
finally
|
||||
{
|
||||
redactor.close();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Equal values keep distinct record bindings; canonical decimals and escaped text
|
||||
** redact exactly and case-sensitively, with no mutation of model-facing objects.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void recordBindingsAndCanonicalVariants() throws Exception
|
||||
{
|
||||
admin();
|
||||
qInstance.withSupplementalMetaData(new ClassifiedValueKeySourceMetaData().withKeySourceInstance(new ClassifiedValueKeySource()
|
||||
{
|
||||
/*******************************************************************************
|
||||
** Deterministic test key, never a production secret.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public Optional<KeyAndId> instanceKey()
|
||||
{
|
||||
return (Optional.of(new KeyAndId(new byte[32], "test")));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** No per-record keys in this instance-key test.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public Optional<KeyAndId> keyForRecordField(String table, String field, Serializable pk, boolean create)
|
||||
{
|
||||
return (Optional.empty());
|
||||
}
|
||||
}));
|
||||
var table = qInstance.getTable("patient");
|
||||
try(TranscriptRedactor redactor = new TranscriptRedactor(qInstance, config.getClassificationSource()))
|
||||
{
|
||||
redactor.observe("patient", "diagnosis", 1, phi, phi);
|
||||
redactor.observe("patient", "diagnosis", 2, phi, phi);
|
||||
String first = ClassifiedValueTokens.forRecordField(table, table.getField("diagnosis"), 1, phi);
|
||||
String second = ClassifiedValueTokens.forRecordField(table, table.getField("diagnosis"), 2, phi);
|
||||
assertThat(first).isNotEqualTo(second);
|
||||
String stored = redactor.payload(JsonUtils.toJson(Map.of("table", "patient", "records", List.of(
|
||||
Map.of("id", 1, "values", Map.of("diagnosis", phi)), Map.of("id", 2, "values", Map.of("diagnosis", phi))))));
|
||||
assertThat(stored).contains(first, second).doesNotContain(phi);
|
||||
assertThat(ClassifiedValueTokens.verifyRecordField(table, table.getField("diagnosis"), 1, phi, first)).isTrue();
|
||||
redactor.observe("patient", "diagnosis", 3, new BigDecimal("123.4500"), "123.4500");
|
||||
String numeric = redactor.redact("123.4500 and 123.45");
|
||||
assertThat(numeric).doesNotContain("123.45").contains("hmac:v1:test:");
|
||||
assertThat(redactor.redact(phi.toLowerCase())).isEqualTo(phi.toLowerCase());
|
||||
redactor.variant("patient", "diagnosis", "1", phi.substring(0, 25));
|
||||
assertThat(redactor.redact(phi.substring(0, 25))).isEqualTo(first);
|
||||
redactor.truncated("Detail: " + pii, "Detail: " + pii.substring(0, 20));
|
||||
redactor.observe("patient", "email", 1, pii, pii);
|
||||
redactor.truncated("Detail: " + pii, "Detail: " + pii.substring(0, 20));
|
||||
assertThat(redactor.redact("Detail: " + pii.substring(0, 20))).doesNotContain("CANARY");
|
||||
assertThat(redactor.redact(first)).isEqualTo(first);
|
||||
redactor.observe("patient", "diagnosis", 4, "quote\"line\n", "quote\"line\n");
|
||||
assertThat(redactor.payload(JsonUtils.toJson(Map.of("text", "quote\"line\n")))).doesNotContain("quote");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** No cloud dependency: both rows use the recording provider.
|
||||
*******************************************************************************/
|
||||
class PatientProviderType implements AssistantProviderType
|
||||
{
|
||||
/*******************************************************************************
|
||||
** Fixture adapter.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public String getType()
|
||||
{
|
||||
return ("patient-test");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fixture adapter.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public String getLabel()
|
||||
{
|
||||
return ("Test");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fixture adapter.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public List<String> getModels()
|
||||
{
|
||||
return (List.of("test-model"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fixture adapter.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public String getDefaultApiKeyEnvName()
|
||||
{
|
||||
return (null);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fixture adapter.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public String getDefaultBaseUrl()
|
||||
{
|
||||
return ("http://localhost");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fixture adapter.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public AssistantModelProvider create(String apiKey, String baseUrl)
|
||||
{
|
||||
return (provider);
|
||||
}
|
||||
|
||||
|
||||
|
||||
}
|
||||
}
|
||||
+513
@@ -0,0 +1,513 @@
|
||||
/*
|
||||
* Copyright 2026 Kingsrook, LLC
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.assistant.jobs;
|
||||
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import com.kingsrook.qbits.assistant.AssistantRuntime;
|
||||
import com.kingsrook.qbits.assistant.api.AssistantApi;
|
||||
import com.kingsrook.qbits.assistant.engine.RunningTurn;
|
||||
import com.kingsrook.qbits.assistant.engine.TurnEvent;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantConversation;
|
||||
import com.kingsrook.qbits.assistant.model.AssistantTurn;
|
||||
import com.kingsrook.qbits.assistant.provider.fake.ScriptedModelProvider;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelEventListener;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelPart;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelProviderException;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelRequest;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelResponse;
|
||||
import com.kingsrook.qbits.assistant.spi.StreamHandle;
|
||||
import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
|
||||
import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
|
||||
import com.kingsrook.qbits.assistant.uiapi.CoreUiApiGateway;
|
||||
import com.kingsrook.qbits.assistant.uiapi.UiApiCall;
|
||||
import com.kingsrook.qbits.assistant.uiapi.UiApiResult;
|
||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.instances.QInstanceEnricher;
|
||||
import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaData;
|
||||
import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaDataProvider;
|
||||
import com.kingsrook.qqq.backend.core.jobs.model.JobStatus;
|
||||
import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
|
||||
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgressSnapshot;
|
||||
import com.kingsrook.qqq.backend.core.jobs.runtime.DurableJobsRuntime;
|
||||
import com.kingsrook.qqq.backend.core.jobs.store.DurableJobStore;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.utils.JsonUtils;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.CsvSource;
|
||||
import static com.kingsrook.qbits.assistant.testing.AssistantTestFixture.EDITOR;
|
||||
import static com.kingsrook.qqq.backend.core.jobs.DurableJobsTables.TABLE_JOB;
|
||||
import static com.kingsrook.qqq.backend.core.jobs.DurableJobsTables.TABLE_LOG;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Real Jobs tray rows, progress and cancellation for assistant answers.
|
||||
*******************************************************************************/
|
||||
class JobTrackingRunTrackerTest extends BaseAssistantTest
|
||||
{
|
||||
private static final String PROMPT = "PRIVATE-PROMPT-JOBS-CANARY";
|
||||
private static final String ANSWER = "PRIVATE-ANSWER-JOBS-CANARY";
|
||||
|
||||
private final CountDownLatch toolEntered = new CountDownLatch(1);
|
||||
private final CountDownLatch toolRelease = new CountDownLatch(1);
|
||||
private DurableJobStore jobs;
|
||||
private AssistantApi api;
|
||||
private ControlledProvider controlled;
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Add jobs only to this fixture, leaving runtime startup explicit per test.
|
||||
*******************************************************************************/
|
||||
@BeforeEach
|
||||
void initializeJobs() throws Exception
|
||||
{
|
||||
new DurableJobsMetaDataProvider().defineAll(qInstance, AssistantTestFixture.BACKEND, null);
|
||||
DurableJobsMetaData settings = new DurableJobsMetaData().withWorkerThreads(1).withLeaseTtlSeconds(3)
|
||||
.withReaperIntervalSeconds(1).withPollIntervalMillis(50).withPodId("test-pod");
|
||||
qInstance.withSupplementalMetaData(settings);
|
||||
new QInstanceEnricher(qInstance).enrich();
|
||||
jobs = new DurableJobStore(settings);
|
||||
|
||||
config.getRuntime().close();
|
||||
config.withUiApiGateway(new CoreUiApiGateway(config::getUiApiBaseUrl, qInstance)
|
||||
{
|
||||
/*******************************************************************************
|
||||
** Hold the real count call so the tool phase can be inspected in storage.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public UiApiResult invoke(UiApiCall call) throws QException
|
||||
{
|
||||
if("/table/{tableName}/count".equals(call.getPathTemplate()))
|
||||
{
|
||||
toolEntered.countDown();
|
||||
try
|
||||
{
|
||||
assertThat(toolRelease.await(15, TimeUnit.SECONDS)).as("tool released").isTrue();
|
||||
}
|
||||
catch(InterruptedException e)
|
||||
{
|
||||
Thread.currentThread().interrupt();
|
||||
throw (new QException("Test tool interrupted", e));
|
||||
}
|
||||
}
|
||||
return (super.invoke(call));
|
||||
}
|
||||
});
|
||||
config.setRuntime(new AssistantRuntime(qInstance, config));
|
||||
var session = AssistantTestFixture.ownerSession(EDITOR, config);
|
||||
QContext.init(qInstance, session);
|
||||
api = new AssistantApi(config.getRuntime(), session);
|
||||
assertThat(config.getRunTracker()).isExactlyInstanceOf(JobTrackingRunTracker.class);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Release every gate before draining both runtimes, even after an assertion.
|
||||
*******************************************************************************/
|
||||
@AfterEach
|
||||
void stopJobs()
|
||||
{
|
||||
toolRelease.countDown();
|
||||
if(controlled != null)
|
||||
{
|
||||
controlled.handles.forEach(handle -> handle.release.countDown());
|
||||
}
|
||||
config.getRuntime().close();
|
||||
DurableJobsRuntime.stop(qInstance, Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A single linked owner job progresses through thinking, tool and writing.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void answerTracksProgressAndCompletes() throws Exception
|
||||
{
|
||||
assertThat(DurableJobsRuntime.start(qInstance)).isNotNull();
|
||||
ControlledHandle thinking = new ControlledHandle(new ModelResponse().withParts(List.of(
|
||||
ModelPart.toolUse("count-1", "count_records", input("table", "blog_post", "filter", filter("AND")))))
|
||||
.withStopReason(ModelResponse.STOP_TOOL_USE), null, false);
|
||||
ControlledHandle writing = answer(false);
|
||||
useProvider(thinking, writing);
|
||||
Integer conversationId = (Integer) api.createConversation(Map.of()).get("id");
|
||||
RunningTurn run = api.startTurn(conversationId, Map.of("text", PROMPT));
|
||||
entered(thinking);
|
||||
QRecord job = onlyJob();
|
||||
Integer jobId = job.getValueInteger("id");
|
||||
assertThat(job.getValueString("processName")).isEqualTo("assistant.turn");
|
||||
assertThat(job.getValueString("label")).isEqualTo("Assistant answer");
|
||||
assertThat(job.getValueString("userId")).isEqualTo(EDITOR);
|
||||
assertThat(job.getValueString("tableName")).isEqualTo(AssistantConversation.TABLE_NAME);
|
||||
assertThat(JsonUtils.toObject(job.getValueString("sourceJson"), Map.class)).containsEntry("recordId", conversationId);
|
||||
assertPhase(jobId, "thinking", null);
|
||||
|
||||
thinking.release.countDown();
|
||||
assertThat(toolEntered.await(15, TimeUnit.SECONDS)).as("tool entered").isTrue();
|
||||
assertPhase(jobId, "tool", "Counting records");
|
||||
toolRelease.countDown();
|
||||
entered(writing);
|
||||
assertPhase(jobId, "writing", null);
|
||||
writing.release.countDown();
|
||||
|
||||
assertThat(completed(run).getStatus()).isEqualTo("COMPLETE");
|
||||
job = onlyJob();
|
||||
assertThat(job.getValueInteger("id")).isEqualTo(jobId);
|
||||
assertThat(job.getValueString("status")).isEqualTo("COMPLETED");
|
||||
assertThat(job.getValueString("progressMessage")).isEqualTo("Answered.");
|
||||
assertThat(job.getValueString("statusReason")).isEqualTo("Answered.");
|
||||
assertThat(run.getLog().after(0, 1)).filteredOn(event -> TurnEvent.TOOL_RESULT.equals(event.getType()))
|
||||
.singleElement().satisfies(event -> assertThat(event.getData()).containsEntry("outcome", "OK"));
|
||||
assertPrivateTextAbsent();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The runtime's periodic tick observes the same row change as Jobs tray Stop.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void jobsTrayStopCancelsAnswer() throws Exception
|
||||
{
|
||||
DurableJobsRuntime.start(qInstance);
|
||||
ControlledHandle writing = answer(false);
|
||||
useProvider(writing);
|
||||
RunningTurn run = startTurn();
|
||||
entered(writing);
|
||||
Integer jobId = onlyJob().getValueInteger("id");
|
||||
jobs.updateJob(jobId, Map.of("status", "CANCELLING"));
|
||||
|
||||
assertThat(completed(run).getStatus()).isEqualTo("CANCELLED");
|
||||
assertThat(writing.isCancelled()).isTrue();
|
||||
assertThat(onlyJob().getValueString("status")).isEqualTo("CANCELLED");
|
||||
assertThat(onlyJob().getValueString("progressMessage")).isEqualTo("Stopped.");
|
||||
assertPrivateTextAbsent();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Even partial output and provider errors containing private text stay out.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void failedAnswerHasOnlyFixedJobMessages() throws Exception
|
||||
{
|
||||
DurableJobsRuntime.start(qInstance);
|
||||
ControlledHandle writing = answer(true);
|
||||
useProvider(writing);
|
||||
RunningTurn run = startTurn();
|
||||
entered(writing);
|
||||
writing.release.countDown();
|
||||
|
||||
assertThat(completed(run).getStatus()).isEqualTo("ERROR");
|
||||
QRecord job = onlyJob();
|
||||
assertThat(job.getValueString("status")).isEqualTo("FAILED");
|
||||
assertThat(job.getValueString("progressMessage")).isEqualTo("The answer did not finish.");
|
||||
assertThat(job.getValueString("statusReason")).isEqualTo("The answer did not finish.");
|
||||
assertThat(jobs.query(TABLE_LOG, new QQueryFilter())).isNotEmpty();
|
||||
assertPrivateTextAbsent();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Hosts without jobs keep ordinary answers and safe null tracker callbacks.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void absentRuntimeDoesNotTrackOrDisruptAnswer() throws Exception
|
||||
{
|
||||
assertThat(DurableJobsRuntime.get(qInstance)).isNull();
|
||||
AssistantRunTracker tracker = config.getRunTracker();
|
||||
assertThat(tracker.start(EDITOR, 1, 1, "ignored")).isNull();
|
||||
tracker.progress(null, "thinking", PROMPT);
|
||||
tracker.onCancel(null, () ->
|
||||
{
|
||||
throw (new AssertionError("No job to cancel"));
|
||||
});
|
||||
tracker.finish(null, "COMPLETE", ANSWER);
|
||||
provider.thenText(ANSWER);
|
||||
RunningTurn run = startTurn();
|
||||
assertThat(completed(run).getStatus()).isEqualTo("COMPLETE");
|
||||
assertThat(jobs.query(TABLE_JOB, new QQueryFilter())).isEmpty();
|
||||
assertThat(jobs.query(TABLE_LOG, new QQueryFilter())).isEmpty();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Core mutates phase objects; every new job must start with fresh phases.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void jobsHaveIndependentPhaseState() throws Exception
|
||||
{
|
||||
DurableJobsRuntime.start(qInstance);
|
||||
AssistantRunTracker tracker = config.getRunTracker();
|
||||
Integer conversationId = (Integer) api.createConversation(Map.of()).get("id");
|
||||
String first = tracker.start(EDITOR, 1, conversationId, PROMPT);
|
||||
tracker.progress(first, "writing", ANSWER);
|
||||
String second = tracker.start(EDITOR, 2, conversationId, PROMPT);
|
||||
QRecord secondJob = jobs.getJob(Integer.valueOf(second));
|
||||
JobProgressSnapshot snapshot = JsonUtils.toObject(secondJob.getValueString("progressJson"), JobProgressSnapshot.class);
|
||||
assertThat(snapshot.getPhases()).extracting(JobPhase::getStatus)
|
||||
.containsExactly(JobPhase.Status.PENDING, JobPhase.Status.PENDING, JobPhase.Status.PENDING);
|
||||
tracker.progress(second, "thinking", PROMPT);
|
||||
tracker.progress(first, "unknown", PROMPT);
|
||||
tracker.progress("missing", "tool", PROMPT);
|
||||
assertPhase(Integer.valueOf(first), "writing", null);
|
||||
assertPhase(Integer.valueOf(second), "thinking", null);
|
||||
tracker.finish(first, "COMPLETE", ANSWER);
|
||||
tracker.finish(second, "COMPLETE", ANSWER);
|
||||
assertPrivateTextAbsent();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Terminal mapping is fixed, including unknown and absent assistant statuses.
|
||||
*******************************************************************************/
|
||||
@ParameterizedTest
|
||||
@CsvSource({ "COMPLETE,COMPLETED,Answered.", "CANCELLED,CANCELLED,Stopped.",
|
||||
"ERROR,FAILED,The answer did not finish.", "unknown,FAILED,The answer did not finish.",
|
||||
",FAILED,The answer did not finish." })
|
||||
void terminalMapping(String status, JobStatus expected, String message)
|
||||
{
|
||||
assertThat(JobTrackingRunTracker.jobStatus(status)).isEqualTo(expected);
|
||||
assertThat(JobTrackingRunTracker.finishMessage(status)).isEqualTo(message);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Use independently controlled model calls without changing the tracker.
|
||||
*******************************************************************************/
|
||||
private void useProvider(ControlledHandle... handles)
|
||||
{
|
||||
controlled = new ControlledProvider(List.of(handles));
|
||||
config.withModelProvider(controlled);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Partial text is emitted before this response is released or fails.
|
||||
*******************************************************************************/
|
||||
private ControlledHandle answer(boolean fail)
|
||||
{
|
||||
return (new ControlledHandle(new ModelResponse().withParts(List.of(ModelPart.text(ANSWER)))
|
||||
.withStopReason(ModelResponse.STOP_END_TURN), ANSWER, fail));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Create a conversation and answer as the fixture's editor.
|
||||
*******************************************************************************/
|
||||
private RunningTurn startTurn() throws Exception
|
||||
{
|
||||
Integer conversationId = (Integer) api.createConversation(Map.of()).get("id");
|
||||
return (api.startTurn(conversationId, Map.of("text", PROMPT)));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Wait until a model call has emitted its optional text and is blocked.
|
||||
*******************************************************************************/
|
||||
private void entered(ControlledHandle handle) throws InterruptedException
|
||||
{
|
||||
assertThat(handle.entered.await(15, TimeUnit.SECONDS)).as("model call entered").isTrue();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Check persisted progress while the answer is still running.
|
||||
*******************************************************************************/
|
||||
private void assertPhase(Integer jobId, String phase, String message) throws QException
|
||||
{
|
||||
QRecord job = jobs.getJob(jobId);
|
||||
assertThat(job.getValueString("status")).isEqualTo("RUNNING");
|
||||
assertThat(job.getValueString("currentPhase")).isEqualTo(phase);
|
||||
assertThat(job.getValueString("progressMessage")).isEqualTo(message);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Assert uniqueness across all jobs, not just those matching this turn.
|
||||
*******************************************************************************/
|
||||
private QRecord onlyJob() throws QException
|
||||
{
|
||||
List<QRecord> rows = jobs.query(TABLE_JOB, new QQueryFilter());
|
||||
assertThat(rows).hasSize(1);
|
||||
return (rows.getFirst());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Inspect every persisted job and log field for both private canaries.
|
||||
*******************************************************************************/
|
||||
private void assertPrivateTextAbsent() throws QException
|
||||
{
|
||||
for(String table : List.of(TABLE_JOB, TABLE_LOG))
|
||||
{
|
||||
assertThat(jobs.query(table, new QQueryFilter())).allSatisfy(row ->
|
||||
assertThat(JsonUtils.toJson(row)).doesNotContain(PROMPT, ANSWER));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Wait for finalization and cleanup before reading the assistant turn.
|
||||
*******************************************************************************/
|
||||
private AssistantTurn completed(RunningTurn run) throws Exception
|
||||
{
|
||||
await(() -> config.getRuntime().getRunningTurn(run.getTurnId()) == null);
|
||||
assertThat(run.getLog().isClosed()).isTrue();
|
||||
return (config.getRuntime().getStore().getTurn(EDITOR, run.getTurnId()));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Bound polling by observable completion rather than fixed timing guesses.
|
||||
*******************************************************************************/
|
||||
private static void await(Callable<Boolean> condition) throws Exception
|
||||
{
|
||||
long deadline = System.nanoTime() + Duration.ofSeconds(15).toNanos();
|
||||
while(!condition.call() && System.nanoTime() < deadline)
|
||||
{
|
||||
Thread.sleep(10);
|
||||
}
|
||||
assertThat(condition.call()).as("condition reached within fifteen seconds").isTrue();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Script model responses with separate gates for thinking and writing.
|
||||
*******************************************************************************/
|
||||
private static class ControlledProvider extends ScriptedModelProvider
|
||||
{
|
||||
private final List<ControlledHandle> handles;
|
||||
private final AtomicInteger next = new AtomicInteger();
|
||||
|
||||
/*******************************************************************************
|
||||
** Retain the scripted handles for teardown and assertions.
|
||||
*******************************************************************************/
|
||||
ControlledProvider(List<ControlledHandle> handles)
|
||||
{
|
||||
this.handles = handles;
|
||||
}
|
||||
|
||||
/*******************************************************************************
|
||||
** Publish partial text, then let the engine wait on the selected handle.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public StreamHandle stream(ModelRequest request, ModelEventListener listener)
|
||||
{
|
||||
ControlledHandle handle = handles.get(next.getAndIncrement());
|
||||
if(handle.text != null)
|
||||
{
|
||||
listener.onTextDelta(handle.text);
|
||||
}
|
||||
return (handle);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A bounded, cancellable model call that can fail after streaming text.
|
||||
*******************************************************************************/
|
||||
private static class ControlledHandle implements StreamHandle
|
||||
{
|
||||
private final ModelResponse response;
|
||||
private final String text;
|
||||
private final boolean fail;
|
||||
private final CountDownLatch entered = new CountDownLatch(1);
|
||||
private final CountDownLatch release = new CountDownLatch(1);
|
||||
private volatile boolean cancelled;
|
||||
|
||||
/*******************************************************************************
|
||||
** Set the eventual response and optional partial-output failure.
|
||||
*******************************************************************************/
|
||||
ControlledHandle(ModelResponse response, String text, boolean fail)
|
||||
{
|
||||
this.response = response;
|
||||
this.text = text;
|
||||
this.fail = fail;
|
||||
}
|
||||
|
||||
/*******************************************************************************
|
||||
** Stop unblocks the engine without test intervention.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public void cancel()
|
||||
{
|
||||
cancelled = true;
|
||||
release.countDown();
|
||||
}
|
||||
|
||||
/*******************************************************************************
|
||||
** Report whether the real cancellation callback reached the provider.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public boolean isCancelled()
|
||||
{
|
||||
return (cancelled);
|
||||
}
|
||||
|
||||
/*******************************************************************************
|
||||
** Wait until the test or the Jobs tray releases this model call.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public ModelResponse await(Duration timeout) throws ModelProviderException, InterruptedException
|
||||
{
|
||||
entered.countDown();
|
||||
if(!release.await(timeout.toMillis(), TimeUnit.MILLISECONDS))
|
||||
{
|
||||
throw (new ModelProviderException(ModelProviderException.Kind.TIMEOUT, null, "Test provider timed out"));
|
||||
}
|
||||
if(cancelled)
|
||||
{
|
||||
return (new ModelResponse().withStopReason(ModelResponse.STOP_CANCELLED));
|
||||
}
|
||||
if(fail)
|
||||
{
|
||||
throw (new ModelProviderException(ModelProviderException.Kind.OTHER, null, PROMPT + " " + ANSWER));
|
||||
}
|
||||
return (response);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -21,6 +21,7 @@ import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import com.kingsrook.qbits.assistant.AssistantQBitConfig;
|
||||
import com.kingsrook.qbits.assistant.engine.PolicyService;
|
||||
import com.kingsrook.qbits.assistant.engine.TurnClassification;
|
||||
@@ -38,6 +39,7 @@ import com.kingsrook.qbits.assistant.spi.AssistantProviderType;
|
||||
import com.kingsrook.qbits.assistant.spi.EmbeddingRequest;
|
||||
import com.kingsrook.qbits.assistant.spi.EmbeddingResponse;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelPart;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelResponse;
|
||||
import com.kingsrook.qbits.assistant.spi.ModelUsage;
|
||||
import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
|
||||
import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
|
||||
@@ -50,12 +52,17 @@ import com.kingsrook.qqq.backend.core.actions.ai.AiFieldMetaData;
|
||||
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.DataClassification;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
|
||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.middleware.javalin.specs.invoke.UiApiCredentials;
|
||||
import com.kingsrook.qqq.backend.core.utils.JsonUtils;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Tag;
|
||||
import org.junit.jupiter.api.Test;
|
||||
@@ -242,6 +249,49 @@ class RagTest extends BaseAssistantTest
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Indexed PHI is re-read live; snippets reach the eligible model but never
|
||||
** assistant rows or audit values.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
@Tag("redteam")
|
||||
void classifiedKnowledgeCanary() throws Exception
|
||||
{
|
||||
String phi = "CANARY-PHI-" + UUID.randomUUID();
|
||||
String pii = "CANARY-PII-" + UUID.randomUUID();
|
||||
var table = qInstance.getTable("blog_post");
|
||||
table.getField("content").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PHI));
|
||||
table.getField("title").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PII));
|
||||
new UpdateAction().execute(new UpdateInput("blog_post").withRecord(new QRecord().withValue("id", 1)
|
||||
.withValue("title", pii).withValue("content", phi)));
|
||||
config.getRuntime().getStore().updateFields(AssistantProvider.TABLE_NAME, new QRecord().withValue("id", embeddingProvider.getId())
|
||||
.withValue("hipaaEligible", true).withValue("isDefault", true));
|
||||
assertThat(config.getRuntime().getRagService().index(source).errors()).isZero();
|
||||
embeddings.thenToolUse("search_knowledge", Map.of("query", "Drain upgrade verify", "limit", 8)).thenRespond(request ->
|
||||
{
|
||||
assertThat(JsonUtils.toJson(request)).contains(phi, pii);
|
||||
return (new ModelResponse().withParts(List.of(ModelPart.text(phi + " " + pii))).withStopReason(ModelResponse.STOP_END_TURN));
|
||||
});
|
||||
assertThat(runner.run(AssistantTestFixture.ADMIN, "Search knowledge").errorCode()).isNull();
|
||||
as(AssistantTestFixture.ADMIN);
|
||||
for(String name : qInstance.getTables().keySet())
|
||||
{
|
||||
if(name.startsWith("assistant_") || name.toLowerCase().startsWith("audit"))
|
||||
{
|
||||
String rows = JsonUtils.toJson(new QueryAction().execute(new QueryInput(name)).getRecords());
|
||||
for(String marker : List.of(phi, pii))
|
||||
{
|
||||
for(int i = 0; i <= marker.length() - 5; i++)
|
||||
{
|
||||
assertThat(rows).doesNotContain(marker.substring(i, i + 5));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Unchanged content performs no embedding; disappeared records are removed.
|
||||
*******************************************************************************/
|
||||
|
||||
+34
@@ -32,6 +32,7 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.CyclicBarrier;
|
||||
import java.util.concurrent.Executors;
|
||||
@@ -56,7 +57,9 @@ import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
|
||||
import com.kingsrook.qbits.assistant.testing.UiApiTestServer;
|
||||
import com.kingsrook.qbits.assistant.uiapi.UiApiCall;
|
||||
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.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
|
||||
@@ -64,12 +67,14 @@ import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutp
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.security.RecordSecurityLock;
|
||||
import com.kingsrook.qqq.backend.core.model.session.QSession;
|
||||
import com.kingsrook.qqq.backend.core.modules.authentication.implementations.MockAuthenticationModule;
|
||||
import com.kingsrook.qqq.backend.core.utils.JsonUtils;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
@@ -213,6 +218,35 @@ class ScheduledTasksTest extends BaseAssistantTest
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Scheduled summaries are derived from the redacted stored message, not live SSE.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void scheduledSummaryCanary() throws Exception
|
||||
{
|
||||
String marker = "CANARY-PII-" + UUID.randomUUID();
|
||||
qInstance.getTable("page").getField("title").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PII));
|
||||
new UpdateAction().execute(new UpdateInput("page").withRecord(new QRecord().withValue("id", 1).withValue("title", marker)));
|
||||
provider.thenToolUse("get_record", Map.of("table", "page", "id", "1")).thenText(marker);
|
||||
AssistantScheduledRun run = scheduledRunner().run(task(), false);
|
||||
assertThat(run.getStatus()).as(run.getError()).isEqualTo("OK");
|
||||
assertThat(run.getSummary()).isEqualTo("changed:v1");
|
||||
assertThat(JsonUtils.toJson(provider.getRequests().getLast())).contains(marker);
|
||||
for(String table : qInstance.getTables().keySet())
|
||||
{
|
||||
if(table.startsWith("assistant_") || table.toLowerCase().startsWith("audit"))
|
||||
{
|
||||
String rows = JsonUtils.toJson(new QueryAction().execute(new QueryInput(table)).getRecords());
|
||||
for(int i = 0; i <= marker.length() - 5; i++)
|
||||
{
|
||||
assertThat(rows).doesNotContain(marker.substring(i, i + 5));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Every tool HTTP call carries both the designated UUID and client channel.
|
||||
*******************************************************************************/
|
||||
|
||||
@@ -173,7 +173,7 @@ public class TurnRunner
|
||||
{
|
||||
last = batch.get(batch.size() - 1).getSeq();
|
||||
}
|
||||
if(batch.isEmpty() && running.getLog().isClosed())
|
||||
if(batch.isEmpty() && running.getLog().isClosed() && running.getLog().size() <= last)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user