diff --git a/.claude/docs/shells.md b/.claude/docs/shells.md index 4945f55a1f1..d9f89d45dd6 100644 --- a/.claude/docs/shells.md +++ b/.claude/docs/shells.md @@ -16,7 +16,7 @@ Gotchas this shell paid for (all confirmed by `IntentBuilderShellIT`): - **Never put an Alpine binding on ``.** The plugin REPLACES the `` with the rendered svg, so a binding would be lost — it throws instead, and the uncaught throw **aborts Alpine's walk over everything below it** (the toolbar rendered, the whole split pane did not). Use an `` placeholder for any `:data-lucide` / `:class` / `x-show`. - **The idempotent creates answer `304`, and `response.ok` is 2xx-only.** `POST /workspaces/{ws}` and `.../{ws}/{project}` return 201 the first time and **304 NOT MODIFIED** afterwards, so a naive `!response.ok` throw makes the *second* save fail while the first succeeded. The fetch helper takes an `alsoOk: [304]`. - **The workspace may not exist at all** on an instance where the user never opened the IDE; creating a project inside a missing workspace answers 404, so `ensureProject` creates the workspace first. -- The agent endpoint can make three upstream calls at 120s each — the client timeout is **7 minutes**, with a staged typing indicator. +- The agent endpoint can make three upstream calls, each **streamed** with adaptive thinking (so there is no fixed per-call ceiling to multiply; the server's outer bound is 10 minutes per call) — the client timeout is **20 minutes**, with a staged typing indicator. - **Ask `GET /services/ide/intent/agent/status` whether the assistant is usable, never a throwaway `/agent` turn.** The shell probes on load so an unconfigured instance says so before the user types; the status endpoint reads configuration only, so it costs no upstream model call (a probe that posted a real turn would burn an Anthropic round-trip on every page load of a *working* instance). The mxGraph rendering is **shared, not copied**: `editor-intent/js/intent-diagrams.js` exposes the framework-free `window.IntentDiagrams.render(model, host)` / `.dispose(host)`, extracted from the Intent Editor's controller and loaded by both (the shell by absolute URL). A standalone page must link **`platform-core/ui/styles/fonts.css`** or the glue cards' SAP-icon glyphs render as tofu. Tests: `IntentBuilderShellIT` — a `@Tag("smoke")` bootstrap check plus the full journey against a **local HTTP stub of the Anthropic upstream** (`DIRIGIBLE_INTENT_AI_BASE_URL`, the same override `IntentAgentServiceTest` uses in-process), so it needs no key and no network. diff --git a/components/engine/engine-intent/CLAUDE.md b/components/engine/engine-intent/CLAUDE.md index f51c5e10f8a..2052a5062b2 100644 --- a/components/engine/engine-intent/CLAUDE.md +++ b/components/engine/engine-intent/CLAUDE.md @@ -259,7 +259,7 @@ All implementations are Spring `@Component` beans implementing `IntentTargetGene The third pane of the Intent Editor (toggled by the toolbar's discussion icon) is a natural-language assistant that edits `app.intent` at the developer's altitude - it **proposes a patch to the intent, never a re-emitted model file**, exactly the "edit shape, not file shape" contract from the design notes above. -- **Server-side bridge, key never leaves the server.** `agent/IntentAgentService` calls the Anthropic Messages API through the shared `ai/ModelClient`; `IntentAgentEndpoint` exposes `POST /services/ide/intent/agent` (body `{yaml, message, history}`, returns `{reply, proposedYaml}`) plus `GET /services/ide/intent/agent/status` → `{configured}` — the cheap "is the assistant usable at all?" probe (`IntentAgentService.isConfigured()`, configuration-only, **never** an upstream call) so a client can say the assistant is unavailable *before* the user's first message instead of after it fails with 412. The Builder shell probes it on load; do not implement that check by posting a throwaway turn, which would cost a real model round-trip on every page load. The whole agent feature is one cohesive `...intent.agent` package (DTO records + exceptions package-private). The API key is read via `DirigibleConfig.INTENT_AI_API_KEY` and is **never** sent to the browser. Config: `DIRIGIBLE_INTENT_AI_API_KEY` (blank → assistant disabled, endpoint returns `412`), `_MODEL` (default `claude-opus-4-8`), `_BASE_URL` (default `https://api.anthropic.com`), `_MAX_TOKENS` (`8192`), `_VERSION` (`2023-06-01`). +- **Server-side bridge, key never leaves the server.** `agent/IntentAgentService` calls the Anthropic Messages API through the shared `ai/ModelClient`; `IntentAgentEndpoint` exposes `POST /services/ide/intent/agent` (body `{yaml, message, history}`, returns `{reply, proposedYaml}`) plus `GET /services/ide/intent/agent/status` → `{configured}` — the cheap "is the assistant usable at all?" probe (`IntentAgentService.isConfigured()`, configuration-only, **never** an upstream call) so a client can say the assistant is unavailable *before* the user's first message instead of after it fails with 412. The Builder shell probes it on load; do not implement that check by posting a throwaway turn, which would cost a real model round-trip on every page load. The whole agent feature is one cohesive `...intent.agent` package (DTO records + exceptions package-private). The API key is read via `DirigibleConfig.INTENT_AI_API_KEY` and is **never** sent to the browser. Config: `DIRIGIBLE_INTENT_AI_API_KEY` (blank → assistant disabled, endpoint returns `412`), `_MODEL` (default `claude-opus-4-8`), `_BASE_URL` (default `https://api.anthropic.com`), `_MAX_TOKENS` (`32768`), `_VERSION` (`2023-06-01`). - **The system prompt is an externalized, reviewable resource.** It lives in `engine-intent/src/main/resources/intent-assistant-guide.md` (loaded from the classpath by `ai/AssistantGuide.load(...)` at class init — fail-fast if missing), not an inline string, so it can be edited as documentation and kept in lockstep with what `IntentParser` enforces. It documents the full schema **including the declarative-glue catalog** (notifications/schedules/integrations/inbound/outbound/rollups) and the trigger `businessKey`/`businessKeyStrategy`, plus the recipient grammar (literal / direct field / one-hop `relation.field`, **no braces** — braces are only for `{…}` interpolation in `subject`/`body`). - **Full file via a forced-available tool, then we diff.** The guide teaches Claude the intent YAML schema + the diff-stability rules (change minimally, preserve key order/comments, append don't reorder). Claude returns the **complete** updated YAML through a single `propose_intent` tool (`{explanation, yaml}`); plain-text replies (no tool call) are clarifying questions/answers. The editor renders the proposal as a Monaco **diff** against the current buffer; **Accept** replaces the buffer (Monaco `setValue` → the existing dirty-tracking + debounced re-parse fire), **Reject** discards. The developer still Saves + Generates as usual - **the agent never writes to disk or runs the generators.** This is why "full proposed YAML + we diff it" was chosen over LLM-authored unified diffs (fragile to apply) or structured edit-ops (lose comments/formatting). - **Every proposal is validated server-side before it reaches the editor (bounded self-correction).** `IntentAgentService.chat` runs `IntentParser.parse` on the proposed YAML; on issues it replays the failed proposal as an assistant turn plus a corrective user turn (the parser's issue strings verbatim + "propose the corrected COMPLETE YAML") and calls the model again, at most `MAX_REPAIR_ROUNDS` (= 2) repair rounds per turn — so the first diff the developer sees is normally already valid, and a stubbornly invalid proposal cannot loop forever. After the last round the proposal is returned anyway with the outstanding issues appended to the reply text (the editor's inline validation still applies on Accept). An un-parseable proposal (YAML syntax error) counts as one issue. The extra upstream calls happen only when the first draft is invalid. `IntentAgentServiceTest` covers all four paths (valid first draft, plain-text reply, one repair round, exhausted rounds) with a scripted upstream — the exact first-user-session failure (`editable` listing a non-displayed field) is the fixture. @@ -274,6 +274,8 @@ There are now **two** assistants, and everything they share lives in one package `...intent.ai` is that shared layer: **`ModelClient`** (the only bridge to the Anthropic Messages API — the five `DIRIGIBLE_INTENT_AI_*` keys, the timeouts, the 412/502 error contract via `AssistantNotConfiguredException` / `AssistantUpstreamException`, the tool-call parsing, and the `messages(history, userTurn)` transcript builder that skips any non-`user`/`assistant` role), **`ProposalRepairLoop`** (the bounded validate-and-repair skeleton described above, parameterised by the tool-input member, the markdown fence, a validator and a repair prompt), **`AssistantGuide`** (fail-fast classpath load of a system prompt) and **`ChatTurn`** (the one transcript DTO both endpoints accept). **There is no second API key, no second HTTP client and no second guide loader** — a new assistant surface is a guide, a tool spec and a validator, nothing more. +**The call is streamed and thinks (#6955).** Three limits used to compound, and they tightened exactly as the authored application grew: one blocking request whose 120-second window had to contain the whole document; no `thinking` parameter at all, which on the configured default model means running with NO thinking (adaptive must be sent explicitly on the 4.7/4.8 family — the old `budget_tokens` form is rejected with a 400); and an 8192-token ceiling, against a tool contract that re-emits the COMPLETE `app.intent` every turn AND every repair round. So: `"stream": true` consumed as server-sent events (`assembleReply` concatenates the `text_delta`s as the answer and re-joins the matching `tool_use` block's `input_json_delta` fragments — matched by the block's **`index`**, never "the last tool_use seen", so a fragment of another block can never corrupt the proposal; thinking deltas, `ping`, and unknown event types are skipped, because the API's versioning policy says new ones may appear), `thinking: {"type": "adaptive"}` sent explicitly with the display default left alone (the reasoning is wanted for the answer's sake, not for anyone to read), and the `_MAX_TOKENS` default raised to 32768 — the env var is unchanged, only the fallback moved. Order mattered: streaming came first, because a larger budget or a longer reasoning pass without it converts a truncated answer into a timed-out one. The 412/502 contract is untouched, **including a mid-stream `error` event** (an overload the API reports inside a 200 response) mapping to the same `AssistantUpstreamException` a non-2xx status does; a proposal cut off mid-JSON says so and names the ceiling, and a `stop_reason: "max_tokens"` is logged, since neither cause is visible from either symptom. The per-request timeout is now an outer bound (10 min), not a window the answer must fit. The **default model is deliberately left alone** — with thinking sent explicitly, reasoning no longer depends on which model is configured. Both assistant surfaces go through this one client, so all three changes reached both. `ModelClientStreamTest` scripts a local SSE upstream (assembly, the request body, the interleaved-block case, mid-stream error, truncation, non-2xx); `IntentBuilderShellIT`'s stub streams its proposal in narrow fragments for the same reason. + `...intent.assist` is the **Workbench assistant** (issue #6724): the same integration standard, applied one altitude *down* — to the hand-written Java the intent boundary hands the developer (a `CalculatedField` action, a `JavaDelegate`, a custom controller). `POST /services/ide/intent/assist` (`{workspace, project, path, source, message, history}` → `{reply, proposedSource, diagnostics}`), `ADMINISTRATOR`/`DEVELOPER`, same 412/502. - **Validation is a real compilation, and it must be batched.** `JavaAssistService` compiles a proposal with `JavaSourceCompiler.compileBatch` **together with every other Java source in the project** — `WorkspaceJavaSources` walks the workspace project for them — because that is exactly how the client-Java runtime compiles it: a `custom/` class exists to use the generated entities and repositories, so a proposal checked alone resolves none of its references and every round would "fail". `JavaAssistServiceTest` pins this by running the same proposal with and without its sibling. The compile is **side-effect-free**: never `JavaLoader.rebuild()` — that swaps the class loader and republishes the whole system. diff --git a/components/engine/engine-intent/src/main/java/org/eclipse/dirigible/components/intent/ai/ModelClient.java b/components/engine/engine-intent/src/main/java/org/eclipse/dirigible/components/intent/ai/ModelClient.java index 297cef42416..b93f25c17f8 100644 --- a/components/engine/engine-intent/src/main/java/org/eclipse/dirigible/components/intent/ai/ModelClient.java +++ b/components/engine/engine-intent/src/main/java/org/eclipse/dirigible/components/intent/ai/ModelClient.java @@ -10,6 +10,7 @@ package org.eclipse.dirigible.components.intent.ai; import java.io.IOException; +import java.io.UncheckedIOException; import java.net.URI; import java.net.URISyntaxException; import java.net.http.HttpClient; @@ -18,9 +19,12 @@ import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; +import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.Stream; import org.apache.commons.lang3.StringUtils; import org.eclipse.dirigible.commons.config.DirigibleConfig; @@ -29,9 +33,9 @@ import org.springframework.stereotype.Component; import com.google.gson.Gson; -import com.google.gson.JsonArray; import com.google.gson.JsonElement; import com.google.gson.JsonObject; +import com.google.gson.JsonParseException; /** * The single bridge from this platform to the Anthropic Messages API. @@ -42,6 +46,14 @@ * contract and the tool-call parsing exist once. The API key lives server-side * ({@link DirigibleConfig#INTENT_AI_API_KEY}) and is never sent to a browser; a blank key disables * every assistant ({@link AssistantNotConfiguredException}). + * + *

+ * The call is streamed ({@code "stream": true}, consumed as server-sent events) and asks for + * adaptive thinking. Both are capacity decisions, not cosmetics: the tool contract re-emits + * the COMPLETE {@code app.intent} on every turn and every repair round, so a real application is + * thousands of output tokens - which a single blocking response with one fixed deadline could not + * hold, and which the model should reason about rather than emit cold. On the configured default + * model omitting {@code thinking} means running with no thinking at all, so it is sent explicitly. */ @Component public class ModelClient { @@ -57,6 +69,18 @@ public class ModelClient { /** The Anthropic Messages path, appended to the configured base URL. */ private static final String MESSAGES_PATH = "/v1/messages"; + /** The only line of a server-sent event this client reads; the payload carries its own type. */ + private static final String DATA_PREFIX = "data:"; + + /** + * An outer bound on the whole exchange - deliberately generous, and deliberately not a window the + * answer has to fit into. The previous non-streamed call had to deliver the entire document within + * 120 seconds, which is the wall this bound replaces: a reasoning pass over a few hundred lines of + * structured YAML routinely outlives it, and a truncated-or-timed-out answer is the same failure to + * the user. + */ + private static final Duration RESPONSE_TIMEOUT = Duration.ofMinutes(10); + private final HttpClient httpClient = HttpClient.newBuilder() .connectTimeout(Duration.ofSeconds(15)) .build(); @@ -120,7 +144,7 @@ public boolean isConfigured() { } /** - * One upstream round-trip. + * One upstream round-trip, streamed. * * @param systemPrompt the assistant's guide * @param messages the conversation turns to send, oldest first @@ -139,19 +163,27 @@ public ModelReply call(String systemPrompt, List> messages, HttpRequest httpRequest = HttpRequest.newBuilder() .uri(messagesEndpoint()) - .timeout(Duration.ofSeconds(120)) + .timeout(RESPONSE_TIMEOUT) .header("content-type", "application/json") + .header("accept", "text/event-stream") .header("x-api-key", apiKey) .header("anthropic-version", DirigibleConfig.INTENT_AI_VERSION.getStringValue()) .POST(HttpRequest.BodyPublishers.ofString(body, StandardCharsets.UTF_8)) .build(); try { - HttpResponse response = httpClient.send(httpRequest, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); - if (response.statusCode() < 200 || response.statusCode() >= 300) { - LOGGER.error("AI assistant upstream call failed with status [{}]: {}", response.statusCode(), response.body()); - throw new AssistantUpstreamException("The AI assistant request failed (HTTP " + response.statusCode() + ")."); + HttpResponse> response = httpClient.send(httpRequest, HttpResponse.BodyHandlers.ofLines()); + try (Stream lines = response.body()) { + if (response.statusCode() < 200 || response.statusCode() >= 300) { + LOGGER.error("AI assistant upstream call failed with status [{}]: {}", response.statusCode(), + lines.collect(Collectors.joining("\n"))); + throw new AssistantUpstreamException("The AI assistant request failed (HTTP " + response.statusCode() + ")."); + } + return assembleReply(lines, tool.name()); } - return parseReply(response.body(), tool.name()); + } catch (JsonParseException ex) { + throw new AssistantUpstreamException("The AI assistant returned an unreadable event stream.", ex); + } catch (UncheckedIOException ex) { + throw new AssistantUpstreamException("The connection to the AI assistant was lost before the answer was complete.", ex); } catch (IOException ex) { throw new AssistantUpstreamException("Could not reach the AI assistant.", ex); } catch (InterruptedException ex) { @@ -201,36 +233,147 @@ private static String unusableBaseUrl(String configured) { + " but is [" + configured + "]."; } - /** Shape the Anthropic Messages request: system prompt, the single proposal tool, and the turns. */ + /** + * Shape the Anthropic Messages request: system prompt, the single proposal tool, and the turns - + * streamed, and with adaptive thinking on. + * + *

+ * {@code thinking} is sent explicitly rather than left to the model's default because on the + * configured default model omitting it means running with no thinking at all. The reply is not + * rendered anywhere, so the display default ({@code omitted}) is left alone - the reasoning is + * wanted for the answer's sake, not for the user to read. + */ private static Map requestBody(String systemPrompt, List> messages, ToolSpec tool) { Map toolBody = Map.of("name", tool.name(), "description", tool.description(), "input_schema", tool.inputSchema()); - return Map.of("model", DirigibleConfig.INTENT_AI_MODEL.getStringValue(), "max_tokens", - DirigibleConfig.INTENT_AI_MAX_TOKENS.getIntValue(), "system", systemPrompt, "tools", List.of(toolBody), "messages", - messages); + Map body = new LinkedHashMap<>(); + body.put("model", DirigibleConfig.INTENT_AI_MODEL.getStringValue()); + body.put("max_tokens", DirigibleConfig.INTENT_AI_MAX_TOKENS.getIntValue()); + body.put("stream", Boolean.TRUE); + body.put("thinking", Map.of("type", "adaptive")); + body.put("system", systemPrompt); + body.put("tools", List.of(toolBody)); + body.put("messages", messages); + return body; } - /** Collect the text blocks as the reply; a matching {@code tool_use} block carries the proposal. */ - private static ModelReply parseReply(String responseBody, String toolName) { - JsonObject root = GSON.fromJson(responseBody, JsonObject.class); - JsonArray content = root.getAsJsonArray("content"); + /** + * Assemble one reply out of the event stream: the text blocks concatenated as the answer, and the + * matching {@code tool_use} block's streamed input-JSON fragments re-joined into the proposal. + * + *

+ * A {@code tool_use} block's {@code input} arrives as {@code input_json_delta} fragments of a + * partial JSON string - the granularity is per fragment, not per member - so the pieces are + * concatenated verbatim and parsed once the stream ends. They are matched by the block's + * {@code index}, so a fragment of some other block can never land in the proposal. Thinking deltas + * and the events that carry no content ({@code message_start}, {@code content_block_stop}, + * {@code ping}, and whatever the API adds next) are skipped: the versioning policy is explicit that + * new event types may appear, so an unknown one is not an error. + */ + private static ModelReply assembleReply(Stream lines, String toolName) { StringBuilder text = new StringBuilder(); - JsonObject toolInput = null; - if (content != null) { - for (JsonElement element : content) { - JsonObject block = element.getAsJsonObject(); - String type = block.has("type") ? block.get("type") - .getAsString() - : ""; - if ("text".equals(type) && block.has("text")) { - text.append(block.get("text") - .getAsString()); - } else if ("tool_use".equals(type) && block.has("name") && toolName.equals(block.get("name") - .getAsString())) { - toolInput = block.getAsJsonObject("input"); + StringBuilder toolJson = new StringBuilder(); + int toolBlockIndex = -1; + Iterator events = lines.iterator(); + while (events.hasNext()) { + JsonObject event = eventData(events.next()); + if (event == null) { + continue; + } + switch (StringUtils.defaultString(member(event, "type"))) { + case "error" -> throw streamError(event); + case "content_block_start" -> { + JsonObject block = event.getAsJsonObject("content_block"); + if (block != null && "tool_use".equals(member(block, "type")) && toolName.equals(member(block, "name"))) { + toolBlockIndex = index(event); + } + } + case "content_block_delta" -> { + JsonObject delta = event.getAsJsonObject("delta"); + String deltaType = delta == null ? "" : StringUtils.defaultString(member(delta, "type")); + if ("text_delta".equals(deltaType)) { + text.append(StringUtils.defaultString(member(delta, "text"))); + } else if ("input_json_delta".equals(deltaType) && toolBlockIndex >= 0 && index(event) == toolBlockIndex) { + toolJson.append(StringUtils.defaultString(member(delta, "partial_json"))); + } } + case "message_delta" -> warnIfTruncated(event); + default -> LOGGER.trace("Skipping AI assistant stream event [{}].", member(event, "type")); } } - return new ModelReply(text.toString(), toolInput); + return new ModelReply(text.toString(), toolBlockIndex >= 0 ? toolInput(toolJson.toString()) : null); + } + + /** + * The JSON payload of one server-sent event line, or {@code null} for every line that carries none + * (the {@code event:} name line, comments, and the blank line between events). + * + *

+ * Server-sent events are line-framed and each Messages API event's payload is one JSON object on a + * single line, so a data line is parsed as it stands rather than re-assembled across lines. + */ + private static JsonObject eventData(String line) { + if (line == null || !line.startsWith(DATA_PREFIX)) { + return null; + } + String payload = line.substring(DATA_PREFIX.length()) + .trim(); + return payload.isEmpty() ? null : GSON.fromJson(payload, JsonObject.class); + } + + /** + * The proposal, re-joined. An empty buffer is a tool called with no arguments - not a failure - but + * an unparseable one is the answer having been cut off mid-JSON, which is worth saying plainly: + * that is what the output ceiling looks like from here. + */ + private static JsonObject toolInput(String accumulated) { + String json = StringUtils.trimToEmpty(accumulated); + if (json.isEmpty()) { + return new JsonObject(); + } + try { + return GSON.fromJson(json, JsonObject.class); + } catch (JsonParseException ex) { + throw new AssistantUpstreamException("The AI assistant's proposal was cut off before it was complete." + + " Raise DIRIGIBLE_INTENT_AI_MAX_TOKENS if this repeats.", ex); + } + } + + /** + * An answer that stopped because it ran out of output tokens. Not fatal by itself - a text-only + * reply is still readable, and a truncated proposal fails validation in the repair loop - but the + * cause is invisible from either symptom, so it is named here. + */ + private static void warnIfTruncated(JsonObject event) { + JsonObject delta = event.getAsJsonObject("delta"); + if (delta != null && "max_tokens".equals(member(delta, "stop_reason"))) { + LOGGER.warn("The AI assistant's answer hit the output ceiling of [{}] tokens and was truncated;" + + " raise DIRIGIBLE_INTENT_AI_MAX_TOKENS.", DirigibleConfig.INTENT_AI_MAX_TOKENS.getIntValue()); + } + } + + /** + * An {@code error} event mid-stream - an overload or a rate limit the API reports inside a 200 + * response. Mapped to the same upstream failure a non-2xx status is, so the endpoints answer 502 + * either way. + */ + private static AssistantUpstreamException streamError(JsonObject event) { + JsonObject error = event.getAsJsonObject("error"); + String type = error == null ? null : member(error, "type"); + LOGGER.error("The AI assistant's event stream carried an error [{}]: {}", type, error == null ? null : member(error, "message")); + return new AssistantUpstreamException( + "The AI assistant request failed (" + StringUtils.defaultIfBlank(type, "stream error") + ")."); + } + + /** A string member, or {@code null} when it is absent or JSON null. */ + private static String member(JsonObject object, String name) { + JsonElement value = object.get(name); + return value == null || value.isJsonNull() ? null : value.getAsString(); + } + + /** A content block's index, or {@code -1} when the event carries none. */ + private static int index(JsonObject event) { + JsonElement value = event.get("index"); + return value == null || value.isJsonNull() ? -1 : value.getAsInt(); } /** diff --git a/components/engine/engine-intent/src/test/java/org/eclipse/dirigible/components/intent/ai/ModelClientStreamTest.java b/components/engine/engine-intent/src/test/java/org/eclipse/dirigible/components/intent/ai/ModelClientStreamTest.java new file mode 100644 index 00000000000..0b5d774609c --- /dev/null +++ b/components/engine/engine-intent/src/test/java/org/eclipse/dirigible/components/intent/ai/ModelClientStreamTest.java @@ -0,0 +1,346 @@ +/* + * Copyright (c) 2010-2026 Eclipse Dirigible contributors + * + * All rights reserved. This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v2.0 which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v20.html + * + * SPDX-FileCopyrightText: Eclipse Dirigible contributors SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.dirigible.components.intent.ai; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.eclipse.dirigible.commons.config.Configuration; +import org.eclipse.dirigible.commons.config.DirigibleConfig; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import com.google.gson.Gson; +import com.google.gson.JsonObject; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; + +/** + * The streamed upstream call: the request the client sends, and the reply it assembles out of the + * server-sent events. + * + *

+ * Both halves are the point of the change. The call is streamed because the tool contract re-emits + * the COMPLETE {@code app.intent} on every turn, which one blocking response with a single deadline + * could not hold; and it asks for adaptive thinking, which on the configured default model has to + * be sent explicitly or the document is emitted with no reasoning at all. Neither is visible from a + * reply that merely arrives, so the request body is asserted alongside the assembly. + * + *

+ * The upstream is a local HTTP server scripted per test - no network, no key. + */ +class ModelClientStreamTest { + + private static final String API_KEY_ENV = "DIRIGIBLE_INTENT_AI_API_KEY"; + private static final String BASE_URL_ENV = "DIRIGIBLE_INTENT_AI_BASE_URL"; + + private static final String TOOL_NAME = "propose_intent"; + private static final Gson GSON = new Gson(); + + /** Pause between flushed events when the upstream trickles its answer. */ + private static final long TRICKLE_PAUSE_MILLIS = 40; + + /** Long enough that its serialized form has to be split across many fragments. */ + private static final String PROPOSED_YAML = """ + name: library + entities: + - name: Member + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: name, type: string, length: 120, required: true } + - { name: notes, type: text } + """; + + private HttpServer upstream; + private volatile String scriptedBody = ""; + private volatile int scriptedStatus = 200; + private volatile boolean scriptedTrickle; + private volatile String sentRequestBody; + + @BeforeEach + void startScriptedUpstream() throws IOException { + upstream = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + upstream.createContext("/v1/messages", this::respond); + upstream.start(); + Configuration.set(API_KEY_ENV, "test-key"); + Configuration.set(BASE_URL_ENV, "http://localhost:" + upstream.getAddress() + .getPort()); + } + + @AfterEach + void stopScriptedUpstream() { + Configuration.remove(API_KEY_ENV); + Configuration.remove(BASE_URL_ENV); + if (upstream != null) { + upstream.stop(0); + } + } + + @Test + void theRequestIsStreamedAndAsksForAdaptiveThinking() { + scriptedBody = event("message_stop", "{\"type\":\"message_stop\"}"); + + call(); + + JsonObject body = GSON.fromJson(sentRequestBody, JsonObject.class); + assertTrue(body.get("stream") + .getAsBoolean(), + "the call must be streamed - a non-streamed one has to deliver the whole document inside one deadline"); + assertEquals("adaptive", body.getAsJsonObject("thinking") + .get("type") + .getAsString(), + "adaptive thinking is sent explicitly; omitting it means no thinking at all on the default model"); + assertEquals(DirigibleConfig.INTENT_AI_MAX_TOKENS.getIntValue(), body.get("max_tokens") + .getAsInt()); + assertEquals("32768", DirigibleConfig.INTENT_AI_MAX_TOKENS.getDefaultValue(), + "the default ceiling has to hold a whole application plus its explanation, not one edit"); + } + + @Test + void theTextAndTheProposalAreAssembledFromTheEventStream() { + String toolInput = GSON.toJson(proposal()); + StringBuilder script = new StringBuilder(); + script.append(event("message_start", "{\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"content\":[]}}")); + script.append(": a comment line no client reads\n\n"); + script.append(event("ping", "{\"type\":\"ping\"}")); + script.append(blockStart(0, "{\"type\":\"text\",\"text\":\"\"}")); + script.append(textDelta(0, "Added a Member ")); + script.append(textDelta(0, "entity.")); + script.append(blockStop(0)); + // A thinking block is streamed on the same connection and must not reach either output. + script.append(blockStart(1, "{\"type\":\"thinking\",\"thinking\":\"\"}")); + script.append(thinkingDelta(1, "The model reasons here.")); + script.append(blockStop(1)); + script.append(blockStart(2, toolUseBlock(TOOL_NAME))); + // Split at an arbitrary width, so fragments land inside strings and inside escape sequences - + // which is what "partial JSON strings" means: the pieces are concatenated verbatim. + for (String fragment : fragments(toolInput, 17)) { + script.append(jsonDelta(2, fragment)); + } + script.append(blockStop(2)); + script.append(event("message_delta", "{\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"}}")); + script.append(event("message_stop", "{\"type\":\"message_stop\"}")); + // An event type this client has never seen: the API's versioning policy says new ones may appear. + script.append(event("something_new", "{\"type\":\"something_new\"}")); + scriptedBody = script.toString(); + + ModelClient.ModelReply reply = call(); + + assertEquals("Added a Member entity.", reply.text()); + assertEquals(PROPOSED_YAML, reply.toolString("yaml"), "the streamed fragments re-join into the proposal byte for byte"); + assertEquals("Added a Member entity.", reply.toolString("explanation")); + } + + @Test + void aTextOnlyAnswerCarriesNoToolInput() { + scriptedBody = blockStart(0, "{\"type\":\"text\",\"text\":\"\"}") + textDelta(0, "Which entity do you mean?") + blockStop(0) + + event("message_stop", "{\"type\":\"message_stop\"}"); + + ModelClient.ModelReply reply = call(); + + assertEquals("Which entity do you mean?", reply.text()); + assertNull(reply.toolInput()); + } + + @Test + void theAnswerIsAssembledAcrossRealChunkBoundaries() { + // The previous client read one buffered response, so a whole document had to be generated inside + // a single 120-second window. This is the shape that replaces it: a chunked response of unknown + // length, flushed event by event with the connection held open between them. It cannot assert the + // old wall is gone in wall-clock terms - it asserts the client no longer needs the answer in one + // piece, which is what the wall was made of. + String toolInput = GSON.toJson(proposal()); + StringBuilder script = new StringBuilder(); + script.append(blockStart(0, "{\"type\":\"text\",\"text\":\"\"}")); + script.append(textDelta(0, "Streaming.")); + script.append(blockStop(0)); + script.append(blockStart(1, toolUseBlock(TOOL_NAME))); + for (String fragment : fragments(toolInput, 29)) { + script.append(jsonDelta(1, fragment)); + } + script.append(blockStop(1)); + script.append(event("message_stop", "{\"type\":\"message_stop\"}")); + scriptedBody = script.toString(); + scriptedTrickle = true; + + ModelClient.ModelReply reply = call(); + + assertEquals("Streaming.", reply.text()); + assertEquals(PROPOSED_YAML, reply.toolString("yaml")); + } + + @Test + void aFragmentOfAnotherBlockNeverLandsInTheProposal() { + // Matching by block index rather than by "the last tool_use block seen": interleaved blocks are + // legal, and a fragment of the wrong one would corrupt the proposal into unparseable JSON. + String toolInput = GSON.toJson(proposal()); + scriptedBody = blockStart(0, toolUseBlock("some_other_tool")) + jsonDelta(0, "{\"garbage\":") + + blockStart(1, toolUseBlock(TOOL_NAME)) + jsonDelta(1, toolInput) + jsonDelta(0, "\"more garbage\"}") + blockStop(1) + + event("message_stop", "{\"type\":\"message_stop\"}"); + + assertEquals(PROPOSED_YAML, call().toolString("yaml")); + } + + @Test + void anErrorEventMidStreamIsAnUpstreamFailure() { + // A 529 the API reports inside a 200 response - the endpoints must answer 502 for it exactly as + // they do for a non-2xx status. + scriptedBody = blockStart(0, "{\"type\":\"text\",\"text\":\"\"}") + textDelta(0, "Working on it") + + event("error", "{\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"Overloaded\"}}"); + + assertThrows(AssistantUpstreamException.class, this::call); + } + + @Test + void aProposalCutOffMidJsonIsAnUpstreamFailure() { + scriptedBody = blockStart(0, toolUseBlock(TOOL_NAME)) + jsonDelta(0, "{\"explanation\": \"Added a Member ent"); + + AssistantUpstreamException failure = assertThrows(AssistantUpstreamException.class, this::call); + assertTrue(failure.getMessage() + .contains("cut off"), + "the message has to name the truncation and the ceiling behind it - the JSON parse failure alone says nothing"); + } + + @Test + void aNonSuccessStatusIsAnUpstreamFailure() { + scriptedStatus = 400; + scriptedBody = "{\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"bad request\"}}"; + + assertThrows(AssistantUpstreamException.class, this::call); + } + + @Test + void withoutAnApiKeyNothingIsSent() { + // Blanked rather than removed: a developer machine may well have the real key in its environment, + // and a runtime value is what outranks it. + Configuration.set(API_KEY_ENV, ""); + + assertThrows(AssistantNotConfiguredException.class, this::call); + assertNull(sentRequestBody); + } + + private ModelClient.ModelReply call() { + Map properties = new LinkedHashMap<>(); + properties.put("explanation", "What changed, in prose."); + properties.put("yaml", "The complete app.intent."); + ModelClient.ToolSpec tool = + new ModelClient.ToolSpec(TOOL_NAME, "Propose the complete intent.", ModelClient.stringSchema(properties)); + return new ModelClient().call("You are an assistant.", List.of(Map.of("role", "user", "content", "Add a notes field")), tool); + } + + private void respond(HttpExchange exchange) throws IOException { + try (InputStream in = exchange.getRequestBody()) { + sentRequestBody = new String(in.readAllBytes(), StandardCharsets.UTF_8); + } + exchange.getResponseHeaders() + .set("Content-Type", scriptedStatus == 200 ? "text/event-stream" : "application/json"); + if (scriptedTrickle) { + trickle(exchange); + return; + } + byte[] payload = scriptedBody.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(scriptedStatus, payload.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(payload); + } + } + + /** + * Answer the way the real upstream does: a chunked response of unknown length, one event flushed at + * a time with the connection held open in between. + */ + private void trickle(HttpExchange exchange) throws IOException { + exchange.sendResponseHeaders(200, 0); + try (OutputStream out = exchange.getResponseBody()) { + for (String event : scriptedBody.split("(?<=\\n\\n)")) { + out.write(event.getBytes(StandardCharsets.UTF_8)); + out.flush(); + try { + Thread.sleep(TRICKLE_PAUSE_MILLIS); + } catch (InterruptedException ex) { + Thread.currentThread() + .interrupt(); + return; + } + } + } + } + + private static Map proposal() { + Map input = new LinkedHashMap<>(); + input.put("explanation", "Added a Member entity."); + input.put("yaml", PROPOSED_YAML); + return input; + } + + /** Split a serialized tool input the way the API does - at arbitrary character boundaries. */ + private static List fragments(String json, int width) { + List fragments = new ArrayList<>(); + for (int start = 0; start < json.length(); start += width) { + fragments.add(json.substring(start, Math.min(start + width, json.length()))); + } + return fragments; + } + + /** One server-sent event: the {@code event:} name line this client ignores, plus its data line. */ + private static String event(String type, String data) { + return "event: " + type + "\ndata: " + data + "\n\n"; + } + + private static String blockStart(int index, String contentBlock) { + return event("content_block_start", + "{\"type\":\"content_block_start\",\"index\":" + index + ",\"content_block\":" + contentBlock + "}"); + } + + private static String blockStop(int index) { + return event("content_block_stop", "{\"type\":\"content_block_stop\",\"index\":" + index + "}"); + } + + private static String toolUseBlock(String name) { + return "{\"type\":\"tool_use\",\"id\":\"toolu_1\",\"name\":\"" + name + "\",\"input\":{}}"; + } + + private static String textDelta(int index, String text) { + return delta(index, "text_delta", "text", text); + } + + private static String jsonDelta(int index, String fragment) { + return delta(index, "input_json_delta", "partial_json", fragment); + } + + private static String thinkingDelta(int index, String thinking) { + return delta(index, "thinking_delta", "thinking", thinking); + } + + /** Built through Gson so the fragment's own quotes and escapes are encoded, not hand-escaped. */ + private static String delta(int index, String deltaType, String member, String value) { + JsonObject delta = new JsonObject(); + delta.addProperty("type", deltaType); + delta.addProperty(member, value); + JsonObject payload = new JsonObject(); + payload.addProperty("type", "content_block_delta"); + payload.addProperty("index", index); + payload.add("delta", delta); + return event("content_block_delta", payload.toString()); + } +} diff --git a/components/resources/resources-builder/src/main/resources/META-INF/dirigible/builder/js/services/intentApi.js b/components/resources/resources-builder/src/main/resources/META-INF/dirigible/builder/js/services/intentApi.js index 0df9cb5b579..252b46676c8 100644 --- a/components/resources/resources-builder/src/main/resources/META-INF/dirigible/builder/js/services/intentApi.js +++ b/components/resources/resources-builder/src/main/resources/META-INF/dirigible/builder/js/services/intentApi.js @@ -20,10 +20,12 @@ const SHELLS = '/services/js/platform-core/extension-services/shells.js?extensionPoints=platform-shells'; const PERSPECTIVES = '/services/js/platform-core/extension-services/perspectives.js?extensionPoints=application-perspectives'; - // The agent endpoint makes up to three upstream model calls at 120s each (the first draft plus two - // server-side repair rounds), so the worst case is around six minutes. Anything less and a legitimately - // slow turn is reported to the user as a failure. - const AGENT_TIMEOUT_MS = 7 * 60 * 1000; + // The agent endpoint makes up to three upstream model calls (the first draft plus two server-side + // repair rounds), each of them streamed and each reasoning over the COMPLETE app.intent it has to + // re-emit - so there is no longer a fixed per-call ceiling to multiply, and the server's own outer + // bound is 10 minutes per call. This is deliberately generous: aborting here reports a turn that is + // still being generated as a failure, and the user has no way to tell the two apart. + const AGENT_TIMEOUT_MS = 20 * 60 * 1000; const DEFAULT_TIMEOUT_MS = 60 * 1000; /** A failed call, carrying the status and the parsed body so callers can act on 412 / 422 / 502. */ diff --git a/modules/commons/commons-config/src/main/java/org/eclipse/dirigible/commons/config/DirigibleConfig.java b/modules/commons/commons-config/src/main/java/org/eclipse/dirigible/commons/config/DirigibleConfig.java index da76c2ca7a8..59b05fa0bf4 100644 --- a/modules/commons/commons-config/src/main/java/org/eclipse/dirigible/commons/config/DirigibleConfig.java +++ b/modules/commons/commons-config/src/main/java/org/eclipse/dirigible/commons/config/DirigibleConfig.java @@ -243,8 +243,14 @@ public enum DirigibleConfig { /** Base URL of the Anthropic-compatible API the Intent assistant calls. */ INTENT_AI_BASE_URL("DIRIGIBLE_INTENT_AI_BASE_URL", "https://api.anthropic.com"), - /** Maximum tokens the Intent assistant may generate in a single proposal. */ - INTENT_AI_MAX_TOKENS("DIRIGIBLE_INTENT_AI_MAX_TOKENS", "8192"), + /** + * Maximum tokens the Intent assistant may generate in a single proposal. The tool contract re-emits + * the COMPLETE {@code app.intent} on every turn and every repair round, so the ceiling has to hold + * a whole application plus its explanation, not one edit - a few hundred lines of this YAML is + * thousands of tokens before the JSON string escaping, and the reasoning pass draws on the same + * budget. + */ + INTENT_AI_MAX_TOKENS("DIRIGIBLE_INTENT_AI_MAX_TOKENS", "32768"), /** Anthropic API version header sent by the Intent assistant. */ INTENT_AI_VERSION("DIRIGIBLE_INTENT_AI_VERSION", "2023-06-01"), diff --git a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/ui/tests/IntentBuilderShellIT.java b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/ui/tests/IntentBuilderShellIT.java index fb00be0bb4d..411746ac531 100644 --- a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/ui/tests/IntentBuilderShellIT.java +++ b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/ui/tests/IntentBuilderShellIT.java @@ -14,7 +14,6 @@ import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; import java.time.Duration; -import java.util.List; import java.util.Map; import org.awaitility.Awaitility; @@ -34,6 +33,7 @@ import com.codeborne.selenide.Condition; import com.codeborne.selenide.Selenide; import com.google.gson.Gson; +import com.google.gson.JsonObject; import com.sun.net.httpserver.HttpExchange; import com.sun.net.httpserver.HttpServer; @@ -79,6 +79,9 @@ public class IntentBuilderShellIT extends UserInterfaceIntegrationTest { - { name: amount, type: decimal, precision: 12, scale: 2 } """; + /** How wide each streamed fragment of the proposal is - narrow, so there are many of them. */ + private static final int FRAGMENT_WIDTH = 23; + /** The project the shell derives from the intent's name - the user never types or sees it. */ private static final String PROJECT = "expenses"; private static final String WORKSPACE_PROJECT = IRepositoryStructure.PATH_USERS + "/admin/workspace/" + PROJECT; @@ -108,19 +111,51 @@ void stopStubbedAssistant() { } } - /** The Messages API shape the agent service parses: one {@code propose_intent} tool-use block. */ + /** + * The Messages API shape the agent service parses: one {@code propose_intent} tool-use block, + * delivered as the server-sent event stream the client asks for ({@code "stream": true}). + * + *

+ * The tool input is split across several {@code input_json_delta} fragments on purpose - a stub + * that sent it in one piece would pass even if the client only ever read the first fragment, which + * is exactly the assembly this journey is meant to exercise end to end. + */ private void respondWithProposal(HttpExchange exchange) throws IOException { - String body = new Gson().toJson(Map.of("content", List.of(Map.of("type", "tool_use", "name", "propose_intent", "input", - Map.of("explanation", "Added an Expense entity with a description and an amount.", "yaml", PROPOSED_INTENT))))); - byte[] payload = body.getBytes(StandardCharsets.UTF_8); + String toolInput = new Gson().toJson( + Map.of("explanation", "Added an Expense entity with a description and an amount.", "yaml", PROPOSED_INTENT)); + StringBuilder stream = new StringBuilder(); + stream.append(event("message_start", "{\"type\":\"message_start\",\"message\":{\"id\":\"msg_stub\",\"content\":[]}}")); + stream.append(event("content_block_start", "{\"type\":\"content_block_start\",\"index\":0,\"content_block\":" + + "{\"type\":\"tool_use\",\"id\":\"toolu_stub\",\"name\":\"propose_intent\",\"input\":{}}}")); + for (int start = 0; start < toolInput.length(); start += FRAGMENT_WIDTH) { + JsonObject delta = new JsonObject(); + delta.addProperty("type", "input_json_delta"); + delta.addProperty("partial_json", toolInput.substring(start, Math.min(start + FRAGMENT_WIDTH, toolInput.length()))); + JsonObject payload = new JsonObject(); + payload.addProperty("type", "content_block_delta"); + payload.addProperty("index", 0); + payload.add("delta", delta); + stream.append(event("content_block_delta", payload.toString())); + } + stream.append(event("content_block_stop", "{\"type\":\"content_block_stop\",\"index\":0}")); + stream.append(event("message_delta", "{\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"}}")); + stream.append(event("message_stop", "{\"type\":\"message_stop\"}")); + + byte[] payload = stream.toString() + .getBytes(StandardCharsets.UTF_8); exchange.getResponseHeaders() - .set("Content-Type", "application/json"); + .set("Content-Type", "text/event-stream"); exchange.sendResponseHeaders(200, payload.length); try (OutputStream out = exchange.getResponseBody()) { out.write(payload); } } + /** One server-sent event: the {@code event:} name line, then its single-line JSON data line. */ + private static String event(String type, String data) { + return "event: " + type + "\ndata: " + data + "\n\n"; + } + @Test @Tag("smoke") void builderShell_loads_and_bootstraps() {