diff --git a/docs/changelog.md b/docs/changelog.md index fd9d4ffc6..6aa021677 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -3,6 +3,60 @@ > **Purpose:** Living document tracking all changes, decisions, and reasoning during implementation. Updated as work progresses for easy reference and review. +--- + +## ๐Ÿ”Ž fix(groups): I17 PR #637 review round 2 (2026-08-08) + +**Repo:** EDDI (`feat/group-i17-shared-artifacts`) + +Two comments triaged: + +- **Announce mutex no longer held across listener callbacks (CodeRabbit Major, accepted):** `announceArtifactChanges` held `artifactAnnounceMutex` through `onArtifactUpdated`, so one slow/backpressured SSE client blocked every other turn's end-of-turn drain. Now the mutex guards only the HANDOFF: exactly one thread at a time is the publisher โ€” it drains under the mutex, releases it, fires the callbacks, and loops for late arrivals; every other thread sees the publisher flag and leaves, its changes guaranteed to ride the publisher's next pass. Write order preserved (single announcer, FIFO queue), no caller ever blocks on a listener. New test drives a write + reentrant announce from INSIDE a callback: published exactly once, in order, nothing stranded, no deadlock. +- **CodeQL log-injection (stale):** raised against the initial commit 6aeba1393; the flagged attach-artifacts log was sanitized in round 1 (74c0acaf7). Reply-only. + +`MemberTurnExecutorTest` (14) + artifact suites green. + +--- + +## ๐Ÿ”Ž fix(groups): I17 PR #637 review round 1 (2026-08-08) + +**Repo:** EDDI (`feat/group-i17-shared-artifacts`) + +All 11 review comments (CodeRabbit ร—9, Copilot/CodeQL ร—2) triaged; every one accepted and fixed: + +- **Meta-schema validation at save time** (`ArtifactValidators.schemaSpecProblem`): `getSchema(spec)` only parses โ€” `{"type":"strng"}` passed and misbehaved at write time. Specs now also validate *as instances* against the bundled 2020-12 meta-schema (no network I/O; degrades to parse-only with a WARN if the bundled resource can't load, rather than rejecting every config). +- **ReDoS bound on REGEX validators**: config-authored pattern ร— 256 KB LLM content could backtrack catastrophically and pin the member turn. `checkRegex` now matches through a deadline-guarded `CharSequence` (500 ms, sampled every 1024 char accesses) and refuses the write on expiry โ€” fails closed, like every other broken-spec path. +- **`[null]` validator entries**: `List.copyOf` NPE'd during config deserialization, preempting `requireValidSpecs`' positional message; now an unmodifiable null-tolerant copy. +- **Artifact event ordering + late writes**: drain+announce now holds a per-conversation mutex (two PARALLEL turns ending together could split the queue and publish v2 before v1), and `executeDiscussion`'s `finally` runs one **final announce pass per leg** so a write accepted by a timed-out member's still-running agent is announced instead of stranded. A write after even that pass keeps the artifact โ€” only its live event is best-effort, by design. +- **`listByGroupConversationId` order**: both backends sort DESC; the interface promises oldest-first. Now re-sorted in Java per the contract. +- **`deleteByGroupConversationId`**: same processed-set/no-progress guard as `deleteAllForUser` โ€” an undislodgeable row is counted once and ends the loop instead of spinning `MAX_ERASURE_PASSES` times and inflating the count. +- **Slack mrkdwn injection**: artifact name/editor id are LLM-authored; `` in a name rendered as a real broadcast. Both fields now `&`/`<`/`>`-escaped. +- **Oversize refusal rounds up** (`Math.ceilDiv`): MAX+1 bytes no longer reads "256 KB is over the 256 KB limit". +- **GDPR cascade Javadoc** now names the shared-artifact step; **CodeQL log injection** at `populateArtifacts` sanitized. + +**Tests:** +7 (meta-schema reject, null-entry positional message, catastrophic-regex deadline, late-write announce pass, single-pass write order, oldest-first sort, no-spin cascade delete). Touched suites 1961 tests โ€” green except the 27 known environmental socket-bound errors (SafeHttpClient/SlackWebApi/Weather/WebScraper), which fail identically on an untouched tree. + +--- + +## ๐Ÿ“„ feat(groups): I17 โ€” shared artifacts (blackboard-lite) (2026-08-08) + +**Repo:** EDDI (`feat/group-i17-shared-artifacts`) + +First Wave 2 queue item from `planning/group-collaboration-NEXT.md` ยง3. Agents can now **co-edit typed documents** instead of only talking: four member tools โ€” `createArtifact`, `readArtifact`, `proposeArtifactUpdate`, `listArtifacts` โ€” gated by a new `artifactConfig` on the group config. + +**Design decisions, per the plan (and the plan's own rejections honored):** + +- **Own collection, never embedded.** `SharedArtifact` + `ISharedArtifactStore`/`SharedArtifactStore` follow `GroupConversationStore`'s single-version runtime-document pattern. The discussion loop's whole-document stale-snapshot persists cannot clobber artifact writes, which is also why โ€” unlike I5's task tools โ€” the artifact tools write **through the store directly**. The live registry is still consulted: membership at assembly (`getForMember`, the caller-supplied-id IDOR guard), liveness at write time, and accepted writes ride a new transient change queue on the live `GroupConversation`. +- **Deterministic CAS-and-retry, explicitly not an LLM fusion arbiter.** The version CAS needed a storage primitive that doesn't exist for numbers: `storeIfFieldEquals(String)` text-compares, which "works" on Postgres (`data->>` renders JSON numbers as text) and **silently never matches on Mongo** (typed BSON equality). New `storeIfFieldEquals(โ€ฆ, long)` overload on `IResourceStorage` + both backends, same no-silent-degrade contract (the default throws). Stale writers get the plan's sentence: *"artifact changed since you read it (now vN); re-read and merge your change."* +- **Declarative validators only.** `JSON_SCHEMA` (new dependency `com.networknt:json-schema-validator` โ€” the victools libraries only *generate* schemas), `REGEX`, `MAX_LENGTH`. Specs hard-fail the config save (`ArtifactValidators.requireValidSpecs` from `AgentGroupStore`, `HitlConfigValidation`'s contract); write-time failures are rejection sentences and the gate fails closed on a broken spec. Content โ‰ค 256 KB. +- **Events without a listener reference:** tools can't fire SSE/Slack events (`ToolAssemblyContext` carries no listener โ€” the structural gap that left I5's planned `task_added_by_agent` unfired). Accepted writes queue an `ArtifactChange` on the live instance; `MemberTurnExecutor` drains the queue in a `finally` after every turn and fires the new `artifact_updated` event (sink constant + record + SSE forward + Slack line + OpenAPI description lists). Drained even with a null listener so the queue cannot grow unbounded. +- **Lifecycle:** artifacts are attached to the discussion status payload at read time in the service (so REST *and* MCP `read_group_conversation` carry them โ€” `availableActions` idiom, `READ_ONLY`, never trusted back from storage); close/delete cascade removes them (`GroupLifecycleOps`, warn-and-continue so a broken artifact store can't make discussions undeletable); GDPR erasure sweeps them **user-keyed** via a stamped `ownerUserId` (page/exact-recheck/fail-loud contract copied from the group store) as a new `GdprComplianceService` cascade step. +- **Caps:** `maxArtifactsPerDiscussion` (default 5) counted inside a `synchronized (liveInstance)` block โ€” creation is check-then-act and PARALLEL phases genuinely race; updates need no lock, the CAS decides. + +**Tests (148 across 8 classes, all green):** tools against a real in-memory CAS store (stale-version retry sentence with the CURRENT version, concurrent same-version writers โ†’ exactly one winner, FINAL freeze, foreign-discussion ids don't resolve, validator chain, refusals leave no side effect); provider gate matrix (every uncertainty โ†’ contribute nothing, membership not existence, `enableBuiltInTools` still applies); store CAS through the numeric overload with `verify(never()).store(โ€ฆ)`; anchored+escaped filters with Java exact-recheck; erasure paging/fail-loud; lifecycle cascade ordering (`inOrder` artifact-delete before document-delete) + cascade-failure-still-deletes; GDPR step + not-resolvable skip + failure-continues; turn-executor drain (exactly once, null-listener drain); Slack lines incl. degenerate-payload skip. **Mutation notes:** degrading the store CAS to an unconditional store does not even compile (the gone-document catch becomes unreachable) โ€” the CAS call is structurally load-bearing; the Mockito-verified negatives (`never().store`, `specs().isEmpty()`) pin the rest. + +**Files:** `SharedArtifact`, `ISharedArtifactStore`, `SharedArtifactStore`, `ArtifactValidators`, `ArtifactTools`, `ArtifactToolsProvider` (+ `AgentOrchestrator` phase-1 wiring), `AgentGroupConfiguration` (`ArtifactConfig`/`ArtifactValidator`/`ValidatorKind`), `GroupConversation` (change queue + read-time `artifacts`), `IResourceStorage` + Mongo/Postgres (numeric CAS), `GroupConversationEventSink`/listener/SSE/Slack, `GroupLifecycleOps`, `GroupConversationService`, `GdprComplianceService`, `AgentGroupStore`, `pom.xml`, `docs/group-conversations.md`, 8 test classes. + --- ## ๐Ÿ”€ merge: bring `origin/main` (PR #627 HITL request pinning) into the branch (2026-08-07) diff --git a/docs/group-conversations.md b/docs/group-conversations.md index 8f440db3b..e01300874 100644 --- a/docs/group-conversations.md +++ b/docs/group-conversations.md @@ -160,6 +160,46 @@ Both caps are enforced independently: `maxPerTurn` bounds a runaway single turn, discussion cap counts only agent-filed tasks, so a large planned backlog does not exhaust it. A rejected call does not consume the per-turn budget. +## Shared artifacts (blackboard-lite) + +Without artifacts, the transcript is the only medium โ€” every structured thing an +agent produces is prose the next agent re-parses. `artifactConfig` gives members +four tools to **create together**: `createArtifact(name, type, content)`, +`readArtifact(nameOrId)`, `proposeArtifactUpdate(nameOrId, content, +expectedVersion, markFinal?)` and `listArtifacts()`. Artifacts are typed +documents (`TEXT`, `MARKDOWN`, `JSON`) in their own collection, listed on the +discussion's REST/MCP status payload as `artifacts`, and announced over SSE and +Slack as `artifact_updated` events. + +```json +"artifactConfig": { + "allowArtifactTools": true, + "maxArtifactsPerDiscussion": 5, + "validators": [ + { "kind": "JSON_SCHEMA", "spec": "{\"type\":\"object\",\"required\":[\"title\"]}" }, + { "kind": "MAX_LENGTH", "spec": "20000" } + ] +} +``` + +**Concurrency is deterministic compare-and-set, not an LLM merge.** Every update +presents the version it read; a stale writer is told *"artifact changed since +you read it (now v3); re-read and merge your change"* and retries against fresh +content. The failure mode is a retry, never a silent bad merge. + +**Validators are declarative only** โ€” `JSON_SCHEMA`, `REGEX` (content must +contain a match), `MAX_LENGTH` (characters) โ€” never code. Specs are checked at +config save time; at write time a failing validator refuses the write with its +message and stores nothing. Content is additionally capped at 256 KB per +artifact. + +Off by default with the same absence discipline as the task tools: no opt-in +means the tools are never assembled. The member agent's own +`enableBuiltInTools` switch still applies. `markFinal: true` freezes an +artifact โ€” FINAL artifacts accept no further updates. Artifacts are deleted +with their discussion (close/delete cascade) and by GDPR erasure; the durable +trace of the work is the transcript. + ## Nested Groups (Group-of-Groups) Members can be other groups. The sub-group runs its own discussion and its synthesized answer becomes the member's response. diff --git a/pom.xml b/pom.xml index 08b6955d8..6700ce946 100644 --- a/pom.xml +++ b/pom.xml @@ -333,6 +333,13 @@ jsonschema-module-jackson 4.38.0 + + + com.networknt + json-schema-validator + 1.5.4 + jakarta.annotation jakarta.annotation-api diff --git a/src/main/java/ai/labs/eddi/configs/groups/ArtifactValidators.java b/src/main/java/ai/labs/eddi/configs/groups/ArtifactValidators.java new file mode 100644 index 000000000..a7f0f27be --- /dev/null +++ b/src/main/java/ai/labs/eddi/configs/groups/ArtifactValidators.java @@ -0,0 +1,289 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.configs.groups; + +import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.ArtifactConfig; +import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.ArtifactValidator; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.networknt.schema.JsonSchema; +import com.networknt.schema.JsonSchemaFactory; +import com.networknt.schema.SchemaId; +import com.networknt.schema.SchemaLocation; +import com.networknt.schema.SpecVersion; +import com.networknt.schema.ValidationMessage; +import org.jboss.logging.Logger; + +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; +import java.util.regex.PatternSyntaxException; +import java.util.stream.Collectors; + +/** + * The declarative artifact validation chain (I17) โ€” JSON Schema, regex and + * length checks a group config can gate artifact writes behind. Declarative + * only: there is deliberately no way to configure code execution here. + *

+ * Two entry points, one per failure audience: {@link #requireValidSpecs} runs + * at config save time and throws so a typo'd spec fails the save (the author's + * problem, at the moment they can fix it); {@link #firstRejection} runs at + * write time and returns a rejection sentence for the model (the write + * is refused, nothing is stored). + * + * @author ginccc + */ +public final class ArtifactValidators { + + /** Bound on schema violation messages quoted back to the model. */ + private static final int MAX_QUOTED_VIOLATIONS = 3; + + /** + * Validation-only mapper: parses candidate content and schema documents, never + * persists anything. + */ + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private static final JsonSchemaFactory SCHEMA_FACTORY = JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.V202012); + + /** + * The 2020-12 meta-schema, resolved from the library's bundled copy (no network + * I/O). {@code getSchema(spec)} alone only parses โ€” it accepts schemas with + * invalid keyword values โ€” so save-time validation additionally validates the + * spec as an instance against this. Null only if the bundled resource + * could not load, in which case save-time validation degrades to parse-only + * rather than rejecting every config. + */ + private static final JsonSchema META_SCHEMA; + + static { + JsonSchema metaSchema = null; + try { + metaSchema = SCHEMA_FACTORY.getSchema(SchemaLocation.of(SchemaId.V202012)); + } catch (Exception e) { + Logger.getLogger(ArtifactValidators.class) + .warnf("2020-12 meta-schema unavailable; schema specs are checked by parse only: %s", e.getMessage()); + } + META_SCHEMA = metaSchema; + } + + /** + * Wall-clock budget for one config-authored regex against one artifact's + * content. A catastrophically backtracking pattern must abort instead of + * pinning the member turn for its full timeout. + */ + private static final long REGEX_DEADLINE_NANOS = TimeUnit.MILLISECONDS.toNanos(500); + + private ArtifactValidators() { + } + + /** + * Save-time spec check: every validator must carry a kind and a spec its kind + * can actually use. Throws {@link IllegalArgumentException} (the same contract + * as {@code HitlConfigValidation}) so the config save fails with an actionable + * message rather than every future artifact write failing at runtime. + */ + public static void requireValidSpecs(ArtifactConfig config) { + if (config == null || config.validators().isEmpty()) { + return; + } + List validators = config.validators(); + for (int i = 0; i < validators.size(); i++) { + ArtifactValidator validator = validators.get(i); + String path = "artifactConfig.validators[" + i + "]"; + if (validator == null || validator.kind() == null) { + throw new IllegalArgumentException(path + " must name a kind (JSON_SCHEMA, REGEX or MAX_LENGTH)"); + } + String spec = validator.spec(); + if (spec == null || spec.isBlank()) { + throw new IllegalArgumentException(path + " (" + validator.kind() + ") needs a spec"); + } + switch (validator.kind()) { + case JSON_SCHEMA -> { + String problem = schemaSpecProblem(spec); + if (problem != null) { + throw new IllegalArgumentException(path + " is not a valid JSON schema: " + problem); + } + } + case REGEX -> { + try { + Pattern.compile(spec); + } catch (PatternSyntaxException e) { + throw new IllegalArgumentException(path + " is not a valid regex: " + e.getMessage()); + } + } + case MAX_LENGTH -> { + int max; + try { + max = Integer.parseInt(spec.trim()); + } catch (NumberFormatException e) { + throw new IllegalArgumentException(path + " (MAX_LENGTH) spec must be a positive integer, not '" + spec + "'"); + } + if (max <= 0) { + throw new IllegalArgumentException(path + " (MAX_LENGTH) must be > 0"); + } + } + } + } + } + + /** + * Write-time gate: runs the chain in config order and returns the first failure + * as a rejection sentence for the model, or {@code null} when every validator + * passes. A broken spec that slipped past save-time validation (hand-edited + * storage) rejects the write rather than admitting it โ€” the gate fails closed. + */ + public static String firstRejection(List validators, String content) { + if (validators == null || validators.isEmpty()) { + return null; + } + String candidate = content != null ? content : ""; + for (ArtifactValidator validator : validators) { + if (validator == null || validator.kind() == null || validator.spec() == null) { + return "This discussion's artifact validation is misconfigured; the write was refused."; + } + String rejection = switch (validator.kind()) { + case MAX_LENGTH -> checkMaxLength(validator.spec(), candidate); + case REGEX -> checkRegex(validator.spec(), candidate); + case JSON_SCHEMA -> checkJsonSchema(validator.spec(), candidate); + }; + if (rejection != null) { + return rejection; + } + } + return null; + } + + private static String checkMaxLength(String spec, String content) { + int max; + try { + max = Integer.parseInt(spec.trim()); + } catch (NumberFormatException e) { + return "This discussion's artifact length validator is misconfigured; the write was refused."; + } + if (content.length() > max) { + return "The content is %d characters, over this discussion's %d-character limit for artifacts. Shorten it." + .formatted(content.length(), max); + } + return null; + } + + private static String checkRegex(String spec, String content) { + try { + if (!Pattern.compile(spec, Pattern.DOTALL).matcher(deadlineGuarded(content)).find()) { + return "The content does not match this discussion's required pattern for artifacts. Expected to find a match of: " + spec; + } + } catch (PatternSyntaxException e) { + return "This discussion's artifact pattern validator is misconfigured; the write was refused."; + } catch (MatchDeadlineExceededException e) { + return "This discussion's artifact pattern validator did not finish in time; the write was refused."; + } + return null; + } + + /** + * Save-time schema check, two layers: the spec must parse, and it must itself + * satisfy the 2020-12 meta-schema โ€” {@code getSchema(spec)} alone accepts e.g. + * {@code {"type": "strng"}} and only misbehaves at write time. Returns the + * problem, or {@code null} for a valid spec. + */ + private static String schemaSpecProblem(String spec) { + JsonNode schemaNode; + try { + schemaNode = MAPPER.readTree(spec); + } catch (Exception e) { + return e.getMessage(); + } + if (META_SCHEMA != null) { + Set violations = META_SCHEMA.validate(schemaNode); + if (!violations.isEmpty()) { + return violations.stream() + .limit(MAX_QUOTED_VIOLATIONS) + .map(ValidationMessage::getMessage) + .collect(Collectors.joining("; ")); + } + } + try { + SCHEMA_FACTORY.getSchema(spec); + } catch (Exception e) { + return e.getMessage(); + } + return null; + } + + /** + * Thrown by {@link #deadlineGuarded}'s wrapper when the match budget is spent. + */ + private static final class MatchDeadlineExceededException extends RuntimeException { + MatchDeadlineExceededException() { + super("regex match exceeded its time budget", null, false, false); + } + } + + /** + * Wraps content so a regex match aborts once {@link #REGEX_DEADLINE_NANOS} is + * spent. Backtracking re-reads characters, so the deadline check in + * {@code charAt} (sampled, to keep the fast path cheap) is hit constantly by + * exactly the pathological patterns it exists to stop. Single-matcher use only + * โ€” the sampling counter is not thread-safe, matching a Matcher's own contract. + */ + private static CharSequence deadlineGuarded(String content) { + long deadline = System.nanoTime() + REGEX_DEADLINE_NANOS; + return new CharSequence() { + private int accesses; + + @Override + public int length() { + return content.length(); + } + + @Override + public char charAt(int index) { + if ((++accesses & 0x3FF) == 0 && System.nanoTime() > deadline) { + throw new MatchDeadlineExceededException(); + } + return content.charAt(index); + } + + @Override + public CharSequence subSequence(int start, int end) { + return content.subSequence(start, end); + } + + @Override + public String toString() { + return content; + } + }; + } + + private static String checkJsonSchema(String spec, String content) { + JsonSchema schema; + try { + schema = SCHEMA_FACTORY.getSchema(spec); + } catch (Exception e) { + return "This discussion's artifact schema validator is misconfigured; the write was refused."; + } + JsonNode node; + try { + node = MAPPER.readTree(content); + } catch (Exception e) { + return "The content must be valid JSON to pass this discussion's artifact schema, and it is not. Fix the JSON and retry."; + } + Set violations = schema.validate(node); + if (!violations.isEmpty()) { + String quoted = violations.stream() + .limit(MAX_QUOTED_VIOLATIONS) + .map(ValidationMessage::getMessage) + .collect(Collectors.joining("; ")); + String suffix = violations.size() > MAX_QUOTED_VIOLATIONS + ? " (and " + (violations.size() - MAX_QUOTED_VIOLATIONS) + " more)" + : ""; + return "The content does not satisfy this discussion's artifact schema: " + quoted + suffix + ". Fix it and retry."; + } + return null; + } +} diff --git a/src/main/java/ai/labs/eddi/configs/groups/ISharedArtifactStore.java b/src/main/java/ai/labs/eddi/configs/groups/ISharedArtifactStore.java new file mode 100644 index 000000000..2f4518904 --- /dev/null +++ b/src/main/java/ai/labs/eddi/configs/groups/ISharedArtifactStore.java @@ -0,0 +1,80 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.configs.groups; + +import ai.labs.eddi.configs.groups.model.SharedArtifact; +import ai.labs.eddi.datastore.IResourceStore; + +import java.util.List; + +/** + * Store for {@link SharedArtifact}s (I17) โ€” runtime documents in their own + * collection, one per co-edited artifact of a group discussion. Same + * single-version, DB-agnostic shape as {@link IGroupConversationStore}. + * + * @author ginccc + */ +public interface ISharedArtifactStore { + + /** + * Persists a new artifact and assigns its id. + * + * @return the new artifact's id + */ + String create(SharedArtifact artifact) throws IResourceStore.ResourceStoreException; + + SharedArtifact read(String id) throws IResourceStore.ResourceNotFoundException, IResourceStore.ResourceStoreException; + + /** + * Persists {@code artifact} only if the stored document's {@code version} still + * equals {@code expectedVersion} โ€” the deterministic CAS every accepted edit + * goes through. The caller applies the edit (which bumps the in-memory version) + * and presents the version it read. + * + * @throws IResourceStore.ResourceModifiedException + * lost the race โ€” someone else's edit landed first (retry after a + * fresh read) + * @throws ArtifactGoneException + * the artifact was deleted concurrently + */ + void updateIfVersion(SharedArtifact artifact, long expectedVersion) + throws IResourceStore.ResourceStoreException, IResourceStore.ResourceModifiedException; + + void delete(String id) throws IResourceStore.ResourceStoreException; + + /** + * All artifacts of one discussion, oldest first. Bounded by the group config's + * {@code maxArtifactsPerDiscussion}, so no pagination surface. + */ + List listByGroupConversationId(String groupConversationId) throws IResourceStore.ResourceStoreException; + + /** + * Cascade delete for a closing/deleted discussion. + * + * @return how many artifacts were removed + */ + long deleteByGroupConversationId(String groupConversationId) throws IResourceStore.ResourceStoreException; + + /** + * GDPR erasure sweep: permanently removes every artifact whose + * {@code ownerUserId} is {@code userId}. Follows the group-conversation store's + * erasure contract โ€” pages until an empty pass, re-checks ownership by exact + * match in Java, and throws rather than reporting a partial erasure as success. + * + * @return how many artifacts were removed + */ + long deleteAllForUser(String userId) throws IResourceStore.ResourceStoreException; + + /** + * Unchecked "deleted concurrently" โ€” thrown by {@link #updateIfVersion} so CAS + * call sites can distinguish a retryable conflict (409) from a gone document + * (404) without a checked-exception cascade. + */ + class ArtifactGoneException extends RuntimeException { + public ArtifactGoneException(String message, Throwable cause) { + super(message, cause); + } + } +} diff --git a/src/main/java/ai/labs/eddi/configs/groups/model/AgentGroupConfiguration.java b/src/main/java/ai/labs/eddi/configs/groups/model/AgentGroupConfiguration.java index 40da7c9d0..0a2593d02 100644 --- a/src/main/java/ai/labs/eddi/configs/groups/model/AgentGroupConfiguration.java +++ b/src/main/java/ai/labs/eddi/configs/groups/model/AgentGroupConfiguration.java @@ -11,6 +11,7 @@ import jakarta.validation.constraints.Size; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Objects; @@ -144,6 +145,73 @@ public GroupTaskConfig() { } } + /** + * Whether members may create and co-edit shared artifacts mid-discussion (I17). + * {@code null} means the artifact tools are absent entirely. + */ + private ArtifactConfig artifactConfig; + + public ArtifactConfig getArtifactConfig() { + return artifactConfig; + } + + public void setArtifactConfig(ArtifactConfig artifactConfig) { + this.artifactConfig = artifactConfig; + } + + /** + * Governs shared artifacts (I17, blackboard-lite): typed documents members + * co-edit through tools instead of re-parsing each other's prose. Same + * opt-in-by-absence discipline as {@link GroupTaskConfig}: off means the tools + * are never assembled, and there is no permissive standalone default. + * + * @param allowArtifactTools + * master switch, default off + * @param maxArtifactsPerDiscussion + * ceiling on artifacts per discussion (default 5). Non-positive + * falls back to the default โ€” 0 must never mean unlimited for an LLM + * write surface + * @param validators + * declarative validation chain every accepted write must pass โ€” + * {@link ValidatorKind#JSON_SCHEMA}, {@link ValidatorKind#REGEX} or + * {@link ValidatorKind#MAX_LENGTH} with a {@code spec}. Declarative + * only, never arbitrary code. Failed validation rejects the + * write with the validator's message; nothing is stored + */ + public record ArtifactConfig(boolean allowArtifactTools, int maxArtifactsPerDiscussion, List validators) { + + public static final int DEFAULT_MAX_ARTIFACTS = 5; + + /** Same normalization choke point as {@link GroupTaskConfig}. */ + public ArtifactConfig { + if (maxArtifactsPerDiscussion <= 0) { + maxArtifactsPerDiscussion = DEFAULT_MAX_ARTIFACTS; + } + // Not List.copyOf: it NPEs on a null ELEMENT ("validators": [null]), + // preempting ArtifactValidators.requireValidSpecs' actionable message. + validators = validators == null ? List.of() : Collections.unmodifiableList(new ArrayList<>(validators)); + } + + /** Disabled, cap at its default, no validators. */ + public ArtifactConfig() { + this(false, DEFAULT_MAX_ARTIFACTS, List.of()); + } + } + + /** + * One declarative artifact validator (I17): {@code kind} selects the check, + * {@code spec} parameterizes it โ€” a JSON schema document, a regex the content + * must match, or a maximum character count. Specs are validated at save time so + * a typo fails the config save, not a member's turn. + */ + public record ArtifactValidator(ValidatorKind kind, String spec) { + } + + /** The closed set of declarative artifact validators (I17). */ + public enum ValidatorKind { + JSON_SCHEMA, REGEX, MAX_LENGTH + } + /** * A member of the group. Members can be individual agents or nested groups. *

diff --git a/src/main/java/ai/labs/eddi/configs/groups/model/GroupConversation.java b/src/main/java/ai/labs/eddi/configs/groups/model/GroupConversation.java index 7e25568af..d37ac9535 100644 --- a/src/main/java/ai/labs/eddi/configs/groups/model/GroupConversation.java +++ b/src/main/java/ai/labs/eddi/configs/groups/model/GroupConversation.java @@ -14,10 +14,12 @@ import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.Map; +import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CopyOnWriteArrayList; /** * Transcript record for a group conversation. Persisted with a single-version @@ -225,6 +227,99 @@ public void setRoundStartTranscriptIndex(int roundStartTranscriptIndex) { @JsonIgnore private transient AgentGroupConfiguration.ProtocolConfig.CostPolicy costCeilingOutcome; + /** + * One accepted artifact write (I17), queued by the artifact tools on the LIVE + * instance and drained by {@code MemberTurnExecutor} after the turn to fire the + * {@code artifact_updated} event. This indirection exists because tools have no + * listener reference โ€” {@code ToolAssemblyContext} carries none โ€” while the + * turn executor does. Transient and concurrent: PARALLEL phases run members + * (and so their tools) concurrently. + */ + public record ArtifactChange(String artifactId, String name, String type, long version, String editorAgentId, + String status, boolean created) { + } + + @JsonIgnore + private final transient Queue pendingArtifactChanges = new ConcurrentLinkedQueue<>(); + + /** + * Serializes the drain HANDOFF over {@link #pendingArtifactChanges}. The queue + * itself is safe, but two PARALLEL turns ending together would split it between + * their drains and could then publish v2's event before v1's. The mutex is held + * only around the drain and the publisher flag โ€” never across the listener + * callbacks, so a slow SSE client cannot block other turns' end-of-turn drains. + * See {@code MemberTurnExecutor#announceArtifactChanges}. + */ + @JsonIgnore + private final transient Object artifactAnnounceMutex = new Object(); + + /** The monitor {@code MemberTurnExecutor} serializes announce passes on. */ + @JsonIgnore + public Object artifactAnnounceMutex() { + return artifactAnnounceMutex; + } + + /** + * Whether a thread is currently PUBLISHING drained artifact changes. Guarded by + * {@link #artifactAnnounceMutex} (never read or written outside it) โ€” this flag + * is what lets the mutex be released during the listener callbacks themselves: + * the active publisher keeps looping over late arrivals, and every other thread + * hands off and leaves instead of blocking on a slow SSE client. Deliberately + * not {@code isX}-named: runtime coordination state, invisible to Jackson. + */ + private transient boolean artifactAnnouncePublishing; + + @JsonIgnore + public boolean artifactAnnouncePublishing() { + return artifactAnnouncePublishing; + } + + @JsonIgnore + public void artifactAnnouncePublishing(boolean publishing) { + this.artifactAnnouncePublishing = publishing; + } + + /** Queues an accepted artifact write for the turn executor to announce. */ + @JsonIgnore + public void queueArtifactChange(ArtifactChange change) { + if (change != null) { + pendingArtifactChanges.add(change); + } + } + + /** Drains queued artifact writes โ€” each drained exactly once. */ + @JsonIgnore + public List drainArtifactChanges() { + List drained = new ArrayList<>(); + ArtifactChange change; + while ((change = pendingArtifactChanges.poll()) != null) { + drained.add(change); + } + return drained; + } + + /** + * The discussion's shared artifacts (I17), populated at READ time by + * {@code GroupConversationService.readGroupConversation} from the artifact + * store โ€” never persisted with this document (artifacts have their own + * collection; see {@link SharedArtifact}). Serialized when populated so REST's + * status payload and MCP's {@code read_group_conversation} both carry it; + * {@code READ_ONLY} because a stored copy must never be trusted back. Mirrors + * the {@code availableActions} idiom. + */ + @JsonIgnore + private transient List artifacts; + + @JsonProperty(value = "artifacts", access = JsonProperty.Access.READ_ONLY) + public List getArtifacts() { + return artifacts; + } + + @JsonIgnore + public void setArtifacts(List artifacts) { + this.artifacts = artifacts; + } + /** * A parent discussion's remaining cost budget at the moment it dispatched this * nested one (I1), or {@code null} for a top-level discussion (and for a parent diff --git a/src/main/java/ai/labs/eddi/configs/groups/model/SharedArtifact.java b/src/main/java/ai/labs/eddi/configs/groups/model/SharedArtifact.java new file mode 100644 index 000000000..627b94994 --- /dev/null +++ b/src/main/java/ai/labs/eddi/configs/groups/model/SharedArtifact.java @@ -0,0 +1,207 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.configs.groups.model; + +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; + +/** + * A typed document a group discussion's members create and co-edit through the + * artifact tools (I17, blackboard-lite) โ€” state that lives outside the + * dialogue, so a structured thing an agent produces stops being prose the next + * agent re-parses. + *

+ * Persisted in its own collection ({@code sharedartifacts}), never + * embedded in the {@link GroupConversation} document: the discussion loop + * persists that document whole from stale snapshots after each phase, which + * would silently clobber concurrent artifact writes. Because artifacts have + * their own collection, tools write them through the store directly โ€” the F1 + * live-instance rule that governs task-list writes does not apply here. + *

+ * Concurrency is deterministic compare-and-set on {@link #version} (see + * {@code ISharedArtifactStore.updateIfVersion}): a stale writer gets a "re-read + * and merge" rejection and retries โ€” explicitly not an LLM fusion + * arbiter, whose failure mode is a silent bad merge rather than a retry. + * + * @author ginccc + */ +public class SharedArtifact { + + /** + * Hard ceiling on {@link #content}, in UTF-8 bytes. An LLM handed a write tool + * can write in a loop; the cap bounds a single document while staying far above + * any deliberated draft. + */ + public static final int MAX_CONTENT_BYTES = 256 * 1024; + + /** + * How many prior revisions ride on the document. A bounded tail, oldest dropped + * โ€” full history is the transcript's job, this exists so a bad edit one or two + * turns back is recoverable without archaeology. + */ + public static final int HISTORY_CAP = 10; + + /** The artifact's content type โ€” how consumers should read {@link #content}. */ + public enum ArtifactType { + TEXT, MARKDOWN, JSON + } + + /** Editing lifecycle: DRAFT while being worked, FINAL once frozen. */ + public enum ArtifactStatus { + DRAFT, FINAL + } + + /** + * One superseded revision. {@code version} is the version this content carried + * while current. + */ + public record ArtifactRevision(String content, String editorAgentId, long version, Instant at) { + } + + private String id; + private String groupConversationId; + /** + * The owning discussion's {@code userId}, stamped at creation so GDPR erasure + * can sweep artifacts by user exactly like group conversations โ€” independent of + * whether the parent document still exists at erasure time. + */ + private String ownerUserId; + private String name; + private ArtifactType type; + private String content; + /** + * Monotonic edit counter, starting at 1 on creation โ€” the CAS token every + * update must present. Serialized as a JSON number; the store's version CAS + * uses the numeric {@code storeIfFieldEquals} overload for exactly that reason. + */ + private long version; + private String lastEditorAgentId; + private ArtifactStatus status = ArtifactStatus.DRAFT; + private List history = new ArrayList<>(); + private Instant createdAt; + private Instant updatedAt; + + /** + * Applies an accepted edit: archives the current content into {@link #history} + * (capped, oldest dropped), then installs the new content and bumps + * {@link #version}. Pure in-memory mutation โ€” persistence and the CAS happen at + * the store. + */ + public void applyEdit(String newContent, String editorAgentId, Instant at) { + history.add(new ArtifactRevision(content, lastEditorAgentId, version, updatedAt != null ? updatedAt : createdAt)); + while (history.size() > HISTORY_CAP) { + history.remove(0); + } + content = newContent; + lastEditorAgentId = editorAgentId; + version = version + 1; + updatedAt = at; + } + + /** + * UTF-8 byte length of {@code content}, for the {@link #MAX_CONTENT_BYTES} cap. + */ + public static int contentBytes(String content) { + return content == null ? 0 : content.getBytes(StandardCharsets.UTF_8).length; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getGroupConversationId() { + return groupConversationId; + } + + public void setGroupConversationId(String groupConversationId) { + this.groupConversationId = groupConversationId; + } + + public String getOwnerUserId() { + return ownerUserId; + } + + public void setOwnerUserId(String ownerUserId) { + this.ownerUserId = ownerUserId; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public ArtifactType getType() { + return type; + } + + public void setType(ArtifactType type) { + this.type = type; + } + + public String getContent() { + return content; + } + + public void setContent(String content) { + this.content = content; + } + + public long getVersion() { + return version; + } + + public void setVersion(long version) { + this.version = version; + } + + public String getLastEditorAgentId() { + return lastEditorAgentId; + } + + public void setLastEditorAgentId(String lastEditorAgentId) { + this.lastEditorAgentId = lastEditorAgentId; + } + + public ArtifactStatus getStatus() { + return status; + } + + public void setStatus(ArtifactStatus status) { + this.status = status; + } + + public List getHistory() { + return history; + } + + public void setHistory(List history) { + this.history = history != null ? history : new ArrayList<>(); + } + + public Instant getCreatedAt() { + return createdAt; + } + + public void setCreatedAt(Instant createdAt) { + this.createdAt = createdAt; + } + + public Instant getUpdatedAt() { + return updatedAt; + } + + public void setUpdatedAt(Instant updatedAt) { + this.updatedAt = updatedAt; + } +} diff --git a/src/main/java/ai/labs/eddi/configs/groups/mongo/AgentGroupStore.java b/src/main/java/ai/labs/eddi/configs/groups/mongo/AgentGroupStore.java index 04ff7972b..d67134a3e 100644 --- a/src/main/java/ai/labs/eddi/configs/groups/mongo/AgentGroupStore.java +++ b/src/main/java/ai/labs/eddi/configs/groups/mongo/AgentGroupStore.java @@ -5,6 +5,7 @@ package ai.labs.eddi.configs.groups.mongo; import ai.labs.eddi.configs.hitl.HitlConfigValidation; +import ai.labs.eddi.configs.groups.ArtifactValidators; import ai.labs.eddi.configs.groups.IAgentGroupStore; import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration; import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.DiscussionPhase; @@ -41,6 +42,7 @@ public AgentGroupStore(IResourceStorageFactory storageFactory, IDocumentBuilder public IResourceStore.IResourceId create(AgentGroupConfiguration groupConfiguration) throws IResourceStore.ResourceStoreException { HitlConfigValidation.validate(groupConfiguration.getHitlConfig()); + ArtifactValidators.requireValidSpecs(groupConfiguration.getArtifactConfig()); normalizeNonPositiveCostCeiling(groupConfiguration); warnOnModeratorlessPhases(groupConfiguration); return super.create(groupConfiguration); @@ -52,6 +54,7 @@ public Integer update(String id, Integer version, AgentGroupConfiguration groupC throws IResourceStore.ResourceStoreException, IResourceStore.ResourceModifiedException, IResourceStore.ResourceNotFoundException { HitlConfigValidation.validate(groupConfiguration.getHitlConfig()); + ArtifactValidators.requireValidSpecs(groupConfiguration.getArtifactConfig()); normalizeNonPositiveCostCeiling(groupConfiguration); warnOnModeratorlessPhases(groupConfiguration); return super.update(id, version, groupConfiguration); diff --git a/src/main/java/ai/labs/eddi/configs/groups/mongo/SharedArtifactStore.java b/src/main/java/ai/labs/eddi/configs/groups/mongo/SharedArtifactStore.java new file mode 100644 index 000000000..eca09940e --- /dev/null +++ b/src/main/java/ai/labs/eddi/configs/groups/mongo/SharedArtifactStore.java @@ -0,0 +1,269 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.configs.groups.mongo; + +import ai.labs.eddi.configs.groups.ISharedArtifactStore; +import ai.labs.eddi.configs.groups.model.SharedArtifact; +import ai.labs.eddi.datastore.IResourceFilter; +import ai.labs.eddi.datastore.IResourceStorage; +import ai.labs.eddi.datastore.IResourceStorageFactory; +import ai.labs.eddi.datastore.IResourceStore; +import ai.labs.eddi.datastore.serialization.IDocumentBuilder; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import org.jboss.logging.Logger; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.regex.Pattern; + +/** + * DB-agnostic store for {@link SharedArtifact}s (I17). Mirrors + * {@link GroupConversationStore}'s single-version runtime-document shape (the + * {@code mongo} package name is historic โ€” this class is backend-neutral via + * {@link IResourceStorageFactory}). + *

+ * The version CAS goes through the numeric + * {@code storeIfFieldEquals(resource, "version", long)} overload: the + * artifact's {@code version} is a JSON number, and MongoDB's typed BSON + * equality would never match it against a string. + * + * @author ginccc + */ +@ApplicationScoped +public class SharedArtifactStore implements ISharedArtifactStore { + + private static final Logger LOGGER = Logger.getLogger(SharedArtifactStore.class); + + private static final int SINGLE_VERSION = 1; + + /** + * Both backends turn a String filter into an UNANCHORED regex โ€” ids must be + * validated before they are interpolated into one. Group-conversation ids are + * Mongo ObjectIds or UUIDs on the two backends, both within this set. + */ + private static final Pattern SAFE_ID = Pattern.compile("[A-Za-z0-9-]+"); + + /** Same spin bound and rationale as {@code GroupConversationStore}. */ + private static final int MAX_ERASURE_PASSES = 1_000; + + private final IResourceStorage storage; + + @Inject + public SharedArtifactStore(IResourceStorageFactory storageFactory, IDocumentBuilder documentBuilder) { + this.storage = storageFactory.create("sharedartifacts", documentBuilder, SharedArtifact.class, + "groupConversationId", "ownerUserId"); + } + + @Override + public String create(SharedArtifact artifact) throws IResourceStore.ResourceStoreException { + try { + IResourceStorage.IResource resource = storage.newResource(artifact); + storage.store(resource); + String id = resource.getId(); + artifact.setId(id); + return id; + } catch (IOException e) { + throw new IResourceStore.ResourceStoreException("Failed to create shared artifact: " + e.getMessage(), e); + } + } + + @Override + public SharedArtifact read(String id) throws IResourceStore.ResourceNotFoundException, IResourceStore.ResourceStoreException { + try { + IResourceStorage.IResource resource = storage.read(id, SINGLE_VERSION); + if (resource == null) { + // Deliberately does not embed the caller-supplied id โ€” reflected-value + // sink, same rule as GroupConversationStore.read. + throw new IResourceStore.ResourceNotFoundException("Shared artifact not found."); + } + SharedArtifact artifact = resource.getData(); + artifact.setId(id); + return artifact; + } catch (IOException e) { + throw new IResourceStore.ResourceStoreException("Failed to read shared artifact: " + e.getMessage(), e); + } + } + + @Override + public void updateIfVersion(SharedArtifact artifact, long expectedVersion) + throws IResourceStore.ResourceStoreException, IResourceStore.ResourceModifiedException { + try { + IResourceStorage.IResource resource = storage.newResource(artifact.getId(), SINGLE_VERSION, artifact); + storage.storeIfFieldEquals(resource, "version", expectedVersion); + } catch (IResourceStore.ResourceNotFoundException e) { + // Unchecked, so CAS call sites can tell "gone" (404) from a genuine + // version conflict (409) โ€” same shape as GroupConversationGoneException. + throw new ArtifactGoneException("Shared artifact no longer exists.", e); + } catch (IOException e) { + throw new IResourceStore.ResourceStoreException("Failed to update shared artifact: " + e.getMessage(), e); + } + } + + @Override + public void delete(String id) throws IResourceStore.ResourceStoreException { + try { + storage.removeAllPermanently(id); + } catch (Exception e) { + throw new IResourceStore.ResourceStoreException("Failed to delete shared artifact: " + e.getMessage(), e); + } + } + + @Override + public List listByGroupConversationId(String groupConversationId) throws IResourceStore.ResourceStoreException { + if (groupConversationId == null || !SAFE_ID.matcher(groupConversationId).matches()) { + return List.of(); + } + try { + var filter = new IResourceFilter.QueryFilters( + List.of(new IResourceFilter.QueryFilter("groupConversationId", "^" + groupConversationId + "$"))); + var resourceIds = storage.findResources( + new IResourceFilter.QueryFilters[]{filter}, "createdAt", 0, IResourceStorage.MAX_RESULT_LIMIT); + List artifacts = new ArrayList<>(); + for (var resourceId : resourceIds) { + try { + var resource = storage.read(resourceId.getId(), SINGLE_VERSION); + if (resource == null) { + continue; + } + SharedArtifact artifact = resource.getData(); + // The anchored regex only narrows; equality decides โ€” same + // narrow-then-recheck rule as every other cross-document filter. + if (!groupConversationId.equals(artifact.getGroupConversationId())) { + continue; + } + artifact.setId(resourceId.getId()); + artifacts.add(artifact); + } catch (IOException e) { + LOGGER.warnf("Skipping unreadable shared artifact %s: %s", resourceId.getId(), e.getMessage()); + } + } + // findResources sorts DESC on both backends; the contract is oldest first. + artifacts.sort(Comparator.comparing(SharedArtifact::getCreatedAt, + Comparator.nullsLast(Comparator.naturalOrder()))); + return artifacts; + } catch (Exception e) { + throw new IResourceStore.ResourceStoreException("Failed to list shared artifacts: " + e.getMessage(), e); + } + } + + @Override + public long deleteByGroupConversationId(String groupConversationId) throws IResourceStore.ResourceStoreException { + if (groupConversationId == null || !SAFE_ID.matcher(groupConversationId).matches()) { + return 0; + } + long deleted = 0; + var processed = new HashSet(); + // maxArtifactsPerDiscussion bounds a discussion's artifacts far below one + // page, but the loop stays honest anyway: delete until a pass finds + // nothing โ€” with the same processed-set/no-progress guard as + // deleteAllForUser, so a row removeAllPermanently cannot dislodge is + // counted once and ends the loop instead of spinning it. + for (int pass = 0; pass < MAX_ERASURE_PASSES; pass++) { + List artifacts = listByGroupConversationId(groupConversationId); + if (artifacts.isEmpty()) { + return deleted; + } + long newThisPass = 0; + for (SharedArtifact artifact : artifacts) { + if (!processed.add(artifact.getId())) { + continue; + } + newThisPass++; + delete(artifact.getId()); + deleted++; + } + if (newThisPass == 0) { + LOGGER.warnf("Cascade delete made no progress; %d shared artifact(s) may remain", artifacts.size()); + return deleted; + } + } + return deleted; + } + + @Override + public long deleteAllForUser(String userId) throws IResourceStore.ResourceStoreException { + if (userId == null || userId.isBlank()) { + return 0; + } + long deleted = 0; + var processed = new HashSet(); + try { + var filter = new IResourceFilter.QueryFilters( + List.of(new IResourceFilter.QueryFilter("ownerUserId", "^" + escapeRegex(userId) + "$"))); + + // Page until an empty pass, always from offset 0 (rows are removed as we + // go); fail loudly on an owned row that will not delete. Contract and + // reasoning identical to GroupConversationStore.deleteAllForUser. + for (int pass = 0; pass < MAX_ERASURE_PASSES; pass++) { + var resourceIds = storage.findResources( + new IResourceFilter.QueryFilters[]{filter}, "createdAt", 0, IResourceStorage.MAX_RESULT_LIMIT); + if (resourceIds == null || resourceIds.isEmpty()) { + return deleted; + } + + long newThisPass = 0; + long failedToDelete = 0; + for (var resourceId : resourceIds) { + if (!processed.add(resourceId.getId())) { + continue; + } + newThisPass++; + try { + var resource = storage.read(resourceId.getId(), SINGLE_VERSION); + if (resource == null) { + continue; + } + if (!userId.equals(resource.getData().getOwnerUserId())) { + // regex matched more than it should have โ€” never delete on it + LOGGER.warnf("Skipping shared artifact %s during erasure: ownerUserId is not an exact match", resourceId.getId()); + continue; + } + storage.removeAllPermanently(resourceId.getId()); + deleted++; + } catch (IOException e) { + failedToDelete++; + LOGGER.warnf("Failed to erase shared artifact %s: %s", resourceId.getId(), e.getMessage()); + } + } + + if (failedToDelete > 0) { + throw new IResourceStore.ResourceStoreException( + "Erasure incomplete: " + failedToDelete + " shared artifact(s) belonging to the user could not be deleted after " + + deleted + " successful deletion(s)"); + } + if (newThisPass == 0) { + return deleted; + } + } + LOGGER.warnf("Erasure stopped after %d passes; more shared artifacts may remain", MAX_ERASURE_PASSES); + } catch (IResourceStore.ResourceStoreException e) { + throw e; + } catch (Exception e) { + throw new IResourceStore.ResourceStoreException("Failed to delete shared artifacts for user: " + e.getMessage(), e); + } + return deleted; + } + + /** + * Backslash-escape the regex metacharacters shared by both backends' engines. + * Not {@link Pattern#quote}: its {@code \Q...\E} form is Java-specific and + * PostgreSQL rejects it. + */ + private static String escapeRegex(String value) { + StringBuilder sb = new StringBuilder(value.length() + 8); + for (int i = 0; i < value.length(); i++) { + char c = value.charAt(i); + if ("\\^$.|?*+()[]{}".indexOf(c) >= 0) { + sb.append('\\'); + } + sb.append(c); + } + return sb.toString(); + } +} diff --git a/src/main/java/ai/labs/eddi/datastore/IResourceStorage.java b/src/main/java/ai/labs/eddi/datastore/IResourceStorage.java index 000d933dc..148d82c03 100644 --- a/src/main/java/ai/labs/eddi/datastore/IResourceStorage.java +++ b/src/main/java/ai/labs/eddi/datastore/IResourceStorage.java @@ -153,6 +153,24 @@ default void storeIfFieldEquals(IResource newResource, String fieldName, Stri + " โ€” a compare-and-swap must never silently degrade to an unconditional store"); } + /** + * As {@link #storeIfFieldEquals(IResource, String, String)}, but comparing a + * JSON number field. A separate overload because the two backends + * disagree about text-comparing numbers: PostgreSQL's {@code data ->> field} + * renders a JSON number as text so {@code "3"} matches, but MongoDB's typed + * BSON equality never matches an int64 against a string โ€” a string-typed CAS on + * a numeric field would "work" on one backend and silently never match on the + * other. Used by the shared-artifact store's version CAS (I17). + *

+ * Same no-fallback contract as the String overload. + */ + default void storeIfFieldEquals(IResource newResource, String fieldName, long expectedValue) + throws IResourceStore.ResourceModifiedException, IResourceStore.ResourceNotFoundException { + throw new UnsupportedOperationException( + "storeIfFieldEquals(long) is not implemented by " + getClass().getName() + + " โ€” a compare-and-swap must never silently degrade to an unconditional store"); + } + /** * Archive {@code history} and apply the version-checked update as ONE unit of * work. diff --git a/src/main/java/ai/labs/eddi/datastore/mongo/MongoResourceStorage.java b/src/main/java/ai/labs/eddi/datastore/mongo/MongoResourceStorage.java index b24fbbe88..240997ead 100644 --- a/src/main/java/ai/labs/eddi/datastore/mongo/MongoResourceStorage.java +++ b/src/main/java/ai/labs/eddi/datastore/mongo/MongoResourceStorage.java @@ -147,6 +147,28 @@ public void storeIfFieldEquals(IResource newResource, String fieldName, Strin } } + @Override + public void storeIfFieldEquals(IResource newResource, String fieldName, long expectedValue) + throws IResourceStore.ResourceModifiedException, IResourceStore.ResourceNotFoundException { + Resource resource = checkInternalResource(newResource); + // Typed BSON equality โ€” the String overload's Filters.eq(field, "3") never + // matches an int64 3, which is exactly why this overload exists. + var result = currentCollection.replaceOne( + Filters.and( + Filters.eq(ID_FIELD, new ObjectId(resource.getId())), + Filters.eq(fieldName, expectedValue)), + resource.getMongoDocument()); + if (result.getMatchedCount() == 0) { + long exists = currentCollection.countDocuments(Filters.eq(ID_FIELD, new ObjectId(resource.getId()))); + if (exists == 0) { + throw new IResourceStore.ResourceNotFoundException( + String.format("Resource no longer exists (id=%s)", resource.getId())); + } + throw new IResourceStore.ResourceModifiedException( + String.format("Resource field '%s' was not %d (id=%s)", fieldName, expectedValue, resource.getId())); + } + } + @Override public void createNew(IResource currentResource) { Resource resource = checkInternalResource(currentResource); diff --git a/src/main/java/ai/labs/eddi/datastore/postgres/PostgresResourceStorage.java b/src/main/java/ai/labs/eddi/datastore/postgres/PostgresResourceStorage.java index 5b6561589..69d58e1ba 100644 --- a/src/main/java/ai/labs/eddi/datastore/postgres/PostgresResourceStorage.java +++ b/src/main/java/ai/labs/eddi/datastore/postgres/PostgresResourceStorage.java @@ -304,6 +304,16 @@ public void storeIfFieldEquals(IResource newResource, String fieldName, Strin } } + @Override + public void storeIfFieldEquals(IResource newResource, String fieldName, long expectedValue) + throws IResourceStore.ResourceModifiedException, IResourceStore.ResourceNotFoundException { + // data ->> field renders a JSON number as its canonical text ("3"), so on + // this backend the numeric CAS is a text comparison against the rendered + // number. The overload exists for MongoDB, whose typed BSON equality has + // no such coincidence โ€” see IResourceStorage's Javadoc. + storeIfFieldEquals(newResource, fieldName, String.valueOf(expectedValue)); + } + @Override public void createNew(IResource resource) { Resource pgResource = checkInternalResource(resource); diff --git a/src/main/java/ai/labs/eddi/engine/api/IGroupConversationService.java b/src/main/java/ai/labs/eddi/engine/api/IGroupConversationService.java index a5a4b77ce..c6a9924d6 100644 --- a/src/main/java/ai/labs/eddi/engine/api/IGroupConversationService.java +++ b/src/main/java/ai/labs/eddi/engine/api/IGroupConversationService.java @@ -210,6 +210,8 @@ default void onConvergenceChecked(GroupConversationEventSink.ConvergenceCheckedE } default void onConvergenceReached(GroupConversationEventSink.ConvergenceReachedEvent event) { } + default void onArtifactUpdated(GroupConversationEventSink.ArtifactUpdatedEvent event) { + } } // --- Exceptions --- diff --git a/src/main/java/ai/labs/eddi/engine/api/IRestGroupConversation.java b/src/main/java/ai/labs/eddi/engine/api/IRestGroupConversation.java index ac078b93b..453bb2dce 100644 --- a/src/main/java/ai/labs/eddi/engine/api/IRestGroupConversation.java +++ b/src/main/java/ai/labs/eddi/engine/api/IRestGroupConversation.java @@ -99,8 +99,8 @@ Response discuss(@PathParam("groupId") String groupId, @NotNull @Operation(summary = "Start a group discussion with SSE streaming", description = "Starts a group discussion asynchronously and streams progress events " + "(group_start, phase_start, speaker_start, speaker_complete, " - + "phase_complete, synthesis_start, group_complete, group_error) " - + "via Server-Sent Events.") + + "phase_complete, synthesis_start, group_complete, group_error, " + + "artifact_updated) via Server-Sent Events.") @APIResponse(responseCode = "200", description = "SSE event stream of discussion progress.") @APIResponse(responseCode = "400", description = "Missing/blank or oversized 'question', or an oversized attachment.") @APIResponse(responseCode = "404", description = "Group not found.") @@ -183,8 +183,8 @@ Response continueDiscussion(@PathParam("groupId") String groupId, description = "Re-run all discussion phases with SSE event streaming for progress. " + "Emits round_start (new round marker) followed by the same events as the " + "initial stream (phase_start, speaker_start, speaker_complete, phase_complete, " - + "synthesis_start, group_complete, group_error), plus the HITL events " - + "(awaiting_approval, hitl_resume, cancelled, member_pause_skipped). NOTE: " + + "synthesis_start, group_complete, group_error, artifact_updated), plus the HITL " + + "events (awaiting_approval, hitl_resume, cancelled, member_pause_skipped). NOTE: " + "'attachments' are NOT supported on a continuation and are rejected with a " + "terminal group_error event rather than silently ignored.") @APIResponse(responseCode = "200", description = "SSE event stream of continuation progress.") diff --git a/src/main/java/ai/labs/eddi/engine/gdpr/GdprComplianceService.java b/src/main/java/ai/labs/eddi/engine/gdpr/GdprComplianceService.java index e4eae91c5..82ccccfec 100644 --- a/src/main/java/ai/labs/eddi/engine/gdpr/GdprComplianceService.java +++ b/src/main/java/ai/labs/eddi/engine/gdpr/GdprComplianceService.java @@ -4,6 +4,7 @@ */ package ai.labs.eddi.engine.gdpr; +import ai.labs.eddi.configs.groups.ISharedArtifactStore; import ai.labs.eddi.configs.groups.mongo.GroupConversationStore; import ai.labs.eddi.configs.properties.IUserMemoryStore; import ai.labs.eddi.configs.properties.model.Property; @@ -62,6 +63,7 @@ public class GdprComplianceService { private final IConversationDescriptorStore conversationDescriptorStore; private final IConversationCheckpointStore checkpointStore; private final Instance groupConversationStoreInstance; + private final Instance sharedArtifactStoreInstance; private final IScheduleStore scheduleStore; private final ICache userConversationCache; @@ -77,6 +79,7 @@ public GdprComplianceService(IUserMemoryStore userMemoryStore, IConversationDescriptorStore conversationDescriptorStore, IConversationCheckpointStore checkpointStore, Instance groupConversationStoreInstance, + Instance sharedArtifactStoreInstance, IScheduleStore scheduleStore, ICacheFactory cacheFactory) { this.userMemoryStore = userMemoryStore; @@ -90,6 +93,7 @@ public GdprComplianceService(IUserMemoryStore userMemoryStore, this.conversationDescriptorStore = conversationDescriptorStore; this.checkpointStore = checkpointStore; this.groupConversationStoreInstance = groupConversationStoreInstance; + this.sharedArtifactStoreInstance = sharedArtifactStoreInstance; this.scheduleStore = scheduleStore; this.userConversationCache = cacheFactory.getCache(USER_CONVERSATION_CACHE_NAME); } @@ -130,6 +134,7 @@ private static String userConversationCacheKey(String intent, String userId) { *

  • Delete all managed conversation mappings (and invalidate their * cache)
    • *
  • Delete all group conversation transcripts
    • + *
  • Delete all shared artifacts owned by the user
    • *
  • Delete all schedules owned by the user
    • *
  • Pseudonymize database log entries
    • *
  • Pseudonymize audit ledger entries
    • @@ -306,6 +311,23 @@ public GdprDeletionResult deleteUserData(String userId) { pseudonym); } + // 5c2. Delete shared artifacts (I17). Artifacts carry ownerUserId (the + // owning discussion's user) precisely so this sweep works even when the + // parent discussion document is already gone. + long sharedArtifactsDeleted = 0; + try { + if (sharedArtifactStoreInstance.isResolvable()) { + sharedArtifactsDeleted = sharedArtifactStoreInstance.get().deleteAllForUser(userId); + if (sharedArtifactsDeleted > 0) { + LOGGER.infof("[GDPR] Deleted %d shared artifacts [%s]", + sharedArtifactsDeleted, pseudonym); + } + } + } catch (Exception e) { + LOGGER.errorf(e, "[GDPR] Failed to delete shared artifacts [%s]", + pseudonym); + } + // 5d. Delete schedules owned by the user. Left behind, they keep firing new // conversations under the erased identity โ€” recreating the data forever. long schedulesDeleted = 0; @@ -346,9 +368,9 @@ public GdprDeletionResult deleteUserData(String userId) { LOGGER.infof("[GDPR] Erasure cascade complete [%s]: " + "memories=%d, conversations=%d, checkpoints=%d, mappings=%d, " - + "groupConversations=%d, schedules=%d, logs=%d, audit=%d", + + "groupConversations=%d, sharedArtifacts=%d, schedules=%d, logs=%d, audit=%d", pseudonym, memoriesDeleted, conversationsDeleted, checkpointsDeleted, - mappingsDeleted, groupConversationsDeleted, schedulesDeleted, + mappingsDeleted, groupConversationsDeleted, sharedArtifactsDeleted, schedulesDeleted, logsPseudonymized, auditPseudonymized); // Write compliance event to immutable audit ledger @@ -360,6 +382,7 @@ public GdprDeletionResult deleteUserData(String userId) { auditDetails.put("conversationsDeleted", conversationsDeleted); auditDetails.put("mappingsDeleted", mappingsDeleted); auditDetails.put("groupConversationsDeleted", groupConversationsDeleted); + auditDetails.put("sharedArtifactsDeleted", sharedArtifactsDeleted); auditDetails.put("schedulesDeleted", schedulesDeleted); auditDetails.put("logsPseudonymized", logsPseudonymized); auditDetails.put("auditPseudonymized", auditPseudonymized); diff --git a/src/main/java/ai/labs/eddi/engine/internal/GroupConversationService.java b/src/main/java/ai/labs/eddi/engine/internal/GroupConversationService.java index a999ec46e..b425d93ad 100644 --- a/src/main/java/ai/labs/eddi/engine/internal/GroupConversationService.java +++ b/src/main/java/ai/labs/eddi/engine/internal/GroupConversationService.java @@ -14,6 +14,7 @@ import ai.labs.eddi.configs.groups.IAgentGroupStore; import ai.labs.eddi.configs.groups.IGroupConversationStore; +import ai.labs.eddi.configs.groups.ISharedArtifactStore; import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration; import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.ContextScope; import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.DiscussionPhase; @@ -208,6 +209,14 @@ public class GroupConversationService implements IGroupConversationService { @Inject LiveDiscussionRegistry liveDiscussionRegistry; + /** + * I17's artifact store. Same field-injection pattern and null-safety as + * {@link #attachmentStore} โ€” {@code null} in direct-construction unit tests, + * where reads simply carry no artifacts and lifecycle cascades no-op. + */ + @Inject + ISharedArtifactStore sharedArtifactStore; + // In-node fast-fail guard for concurrent post-discussion operations // (follow-up, continue, close) on the same conversation: a second // operation on the same gcId is rejected rather than queued. The Set is @@ -1083,6 +1092,13 @@ public GroupConversation executeDiscussion(GroupConversation gc, AgentGroupConfi throw new GroupExecutionException("Group discussion failed: " + e.getMessage(), e); } finally { timerGroupDiscussion.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); + // I17: last announce pass for this leg. A member turn that timed out + // drained an empty queue in ITS finally, while its still-running agent + // could accept an artifact write afterwards; without this, that write's + // event is stranded until (unless) another turn runs. A write accepted + // after even this pass keeps the artifact (the store write already + // committed) โ€” only its live event is best-effort, by design. + MemberTurnExecutor.announceArtifactChanges(gc, listener); // I1: fold this leg's spend into the lifetime gauge. Recorded as a delta // against what this leg started with, so a resumed leg (whose gc arrives // already carrying the pre-pause total) contributes only what it newly @@ -1188,7 +1204,30 @@ private void scheduleGroupHitlTimeout(GroupConversation gc) { @Override public GroupConversation readGroupConversation(String groupConversationId) throws IResourceStore.ResourceNotFoundException, IResourceStore.ResourceStoreException { - return lifecycleOps().readGroupConversation(groupConversationId); + GroupConversation gc = lifecycleOps().readGroupConversation(groupConversationId); + populateArtifacts(gc); + return gc; + } + + /** + * I17: attaches the discussion's shared artifacts as a read-time derived field, + * here in the service so REST's status payload and MCP's + * {@code read_group_conversation} both carry them. Best-effort โ€” a status read + * must not fail because the artifact store hiccuped. + */ + private void populateArtifacts(GroupConversation gc) { + if (sharedArtifactStore == null || gc == null || gc.getId() == null) { + return; + } + try { + var artifacts = sharedArtifactStore.listByGroupConversationId(gc.getId()); + if (!artifacts.isEmpty()) { + gc.setArtifacts(artifacts); + } + } catch (Exception e) { + LOGGER.warnf("Could not attach shared artifacts to group conversation %s: %s", + LogSanitizer.sanitize(gc.getId()), LogSanitizer.sanitize(e.getMessage())); + } } @Override @@ -1242,7 +1281,7 @@ private void failConversation(GroupConversation gc) { private GroupLifecycleOps lifecycleOps() { return new GroupLifecycleOps(conversationStore, groupStore, conversationService, agentFactory, agentStore, - deploymentStore, operationsInProgress, activeTokens, this, + deploymentStore, sharedArtifactStore, operationsInProgress, activeTokens, this, counterGroupFollowUp, counterGroupContinue, counterGroupClose, counterGroupFailure); } diff --git a/src/main/java/ai/labs/eddi/engine/internal/RestGroupConversation.java b/src/main/java/ai/labs/eddi/engine/internal/RestGroupConversation.java index fb54f43b6..9f6c74249 100644 --- a/src/main/java/ai/labs/eddi/engine/internal/RestGroupConversation.java +++ b/src/main/java/ai/labs/eddi/engine/internal/RestGroupConversation.java @@ -875,6 +875,12 @@ public void onDecisionReached(GroupConversationEventSink.DecisionReachedEvent ev // debate verdict before a later synthesis phase). sendEvent(eventSink, sse, GroupConversationEventSink.EVENT_DECISION_REACHED, toJson(event)); } + + @Override + public void onArtifactUpdated(GroupConversationEventSink.ArtifactUpdatedEvent event) { + // Not terminal โ€” artifacts are edited throughout the discussion. + sendEvent(eventSink, sse, GroupConversationEventSink.EVENT_ARTIFACT_UPDATED, toJson(event)); + } }; } diff --git a/src/main/java/ai/labs/eddi/engine/internal/groups/GroupLifecycleOps.java b/src/main/java/ai/labs/eddi/engine/internal/groups/GroupLifecycleOps.java index f02879588..10662bc14 100644 --- a/src/main/java/ai/labs/eddi/engine/internal/groups/GroupLifecycleOps.java +++ b/src/main/java/ai/labs/eddi/engine/internal/groups/GroupLifecycleOps.java @@ -8,6 +8,7 @@ import ai.labs.eddi.configs.deployment.IDeploymentStore; import ai.labs.eddi.configs.groups.IAgentGroupStore; import ai.labs.eddi.configs.groups.IGroupConversationStore; +import ai.labs.eddi.configs.groups.ISharedArtifactStore; import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration; import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.DiscussionPhase; import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.LifecyclePolicy; @@ -83,6 +84,7 @@ public class GroupLifecycleOps { private final IAgentFactory agentFactory; private final IAgentStore agentStore; private final IDeploymentStore deploymentStore; + private final ISharedArtifactStore sharedArtifactStore; private final Set operationsInProgress; private final ConcurrentHashMap activeTokens; private final GroupConversationService groupConversationService; @@ -93,7 +95,7 @@ public class GroupLifecycleOps { public GroupLifecycleOps(IGroupConversationStore conversationStore, IAgentGroupStore groupStore, IConversationService conversationService, IAgentFactory agentFactory, IAgentStore agentStore, - IDeploymentStore deploymentStore, Set operationsInProgress, + IDeploymentStore deploymentStore, ISharedArtifactStore sharedArtifactStore, Set operationsInProgress, ConcurrentHashMap activeTokens, GroupConversationService groupConversationService, Counter counterGroupFollowUp, Counter counterGroupContinue, Counter counterGroupClose, Counter counterGroupFailure) { @@ -103,6 +105,7 @@ public GroupLifecycleOps(IGroupConversationStore conversationStore, IAgentGroupS this.agentFactory = agentFactory; this.agentStore = agentStore; this.deploymentStore = deploymentStore; + this.sharedArtifactStore = sharedArtifactStore; this.operationsInProgress = operationsInProgress; this.activeTokens = activeTokens; this.groupConversationService = groupConversationService; @@ -153,6 +156,10 @@ public void deleteGroupConversation(String groupConversationId) // operations. Delete is terminal, so reclaim any dynamically-created agents // here; otherwise deleting a COMPLETED conversation would orphan them. cleanupEphemeralAgentsForGroup(gc); + // I17: artifacts live in their own collection keyed by this conversation + // โ€” remove them with their discussion, before the document goes (while + // the id is still provably a discussion the caller could delete). + deleteArtifactsForGroupConversation(groupConversationId); conversationStore.delete(groupConversationId); } catch (IResourceStore.ResourceNotFoundException e) { LOGGER.warnf("Group conversation %s not found for deletion", LogSanitizer.sanitize(groupConversationId)); @@ -476,6 +483,11 @@ public GroupConversation closeGroupConversation(String groupConversationId) // Ephemeral agent cleanup (deferred from executeDiscussion) cleanupEphemeralAgentsForGroup(gc); + // I17: close is a lifecycle end โ€” the working artifacts go with it. + // Their durable trace is the transcript (accepted updates are announced + // there and a synthesis quotes what mattered), not the working copies. + deleteArtifactsForGroupConversation(groupConversationId); + LOGGER.infof("Group conversation %s closed โ€” member conversations ended, ephemeral agents cleaned up", LogSanitizer.sanitize(groupConversationId)); @@ -486,6 +498,27 @@ public GroupConversation closeGroupConversation(String groupConversationId) } } + /** + * I17 cascade: removes the discussion's shared artifacts. Warn-and-continue on + * failure โ€” a broken artifact store must not make discussions undeletable; the + * user-keyed GDPR erasure sweep (which fails loudly) is the completeness + * guarantee, this is the tidy path. + */ + private void deleteArtifactsForGroupConversation(String groupConversationId) { + if (sharedArtifactStore == null) { + return; + } + try { + long removed = sharedArtifactStore.deleteByGroupConversationId(groupConversationId); + if (removed > 0) { + LOGGER.infof("Removed %d shared artifact(s) of group conversation %s", removed, LogSanitizer.sanitize(groupConversationId)); + } + } catch (Exception e) { + LOGGER.warnf("Failed to remove shared artifacts of group conversation %s: %s", + LogSanitizer.sanitize(groupConversationId), e.getMessage()); + } + } + public List listGroupPendingApprovals(String groupId, int limit) throws IResourceStore.ResourceStoreException { // Bounded summaries โ€” never hand full transcripts to a listing endpoint. diff --git a/src/main/java/ai/labs/eddi/engine/internal/groups/MemberTurnExecutor.java b/src/main/java/ai/labs/eddi/engine/internal/groups/MemberTurnExecutor.java index aba17f729..f79de4718 100644 --- a/src/main/java/ai/labs/eddi/engine/internal/groups/MemberTurnExecutor.java +++ b/src/main/java/ai/labs/eddi/engine/internal/groups/MemberTurnExecutor.java @@ -152,6 +152,72 @@ public TranscriptEntry executeAgentTurn(GroupMember member, GroupConversation gc DiscussionPhase phase, String targetAgentId, GroupDiscussionEventListener listener, GroupConversationService.MemberTurnCancellation cancellation, String conversationKey) throws GroupDiscussionException { + try { + return doExecuteAgentTurn(member, gc, input, protocol, phaseIdx, phase, targetAgentId, listener, cancellation, conversationKey); + } finally { + // I17: announce artifact writes this turn made. In a finally because an + // accepted write already happened whatever the turn's own outcome โ€” + // a timeout or cancellation after the write must not swallow the event. + // Drained even with a null listener so the queue never grows unbounded. + announceArtifactChanges(gc, listener); + } + } + + /** + * I17: fires {@code artifact_updated} for every write queued during the turn. + * Public (not just the per-turn finally) because the discussion loop calls it + * once more when the leg ends, so a write accepted by a timed-out member's + * still-running agent โ€” whose own turn already drained โ€” is announced rather + * than stranded in the queue. + *

    + * The mutex guards the HANDOFF, never the callbacks. PARALLEL turns + * ending together must not split the queue between their drains and publish + * v2's event before v1's โ€” but holding the monitor across + * {@code listener.onArtifactUpdated} would let one slow, backpressured SSE + * client block every other turn's end-of-turn drain. So exactly one thread at a + * time is the publisher: it drains under the mutex, releases it, fires + * the callbacks, and loops for anything that arrived meanwhile; every other + * thread sees the publisher flag and leaves, its changes guaranteed to ride the + * publisher's next loop. Write order is preserved (single announcer, FIFO + * queue) and no caller ever blocks on a listener. + */ + public static void announceArtifactChanges(GroupConversation gc, GroupDiscussionEventListener listener) { + while (true) { + List changes; + synchronized (gc.artifactAnnounceMutex()) { + if (gc.artifactAnnouncePublishing()) { + // The active publisher's next loop iteration drains our + // changes โ€” leaving keeps this thread off the slow path. + return; + } + changes = gc.drainArtifactChanges(); + if (changes.isEmpty()) { + return; + } + gc.artifactAnnouncePublishing(true); + } + try { + if (listener != null) { + for (GroupConversation.ArtifactChange change : changes) { + listener.onArtifactUpdated(new GroupConversationEventSink.ArtifactUpdatedEvent( + change.artifactId(), change.name(), change.type(), change.version(), + change.editorAgentId(), change.status(), change.created())); + } + } + } finally { + synchronized (gc.artifactAnnounceMutex()) { + gc.artifactAnnouncePublishing(false); + } + } + // Loop: drain-and-publish anything queued while the callbacks ran โ€” + // a thread that handed off above relies on exactly this pass. + } + } + + private TranscriptEntry doExecuteAgentTurn(GroupMember member, GroupConversation gc, String input, ProtocolConfig protocol, int phaseIdx, + DiscussionPhase phase, String targetAgentId, GroupDiscussionEventListener listener, + GroupConversationService.MemberTurnCancellation cancellation, String conversationKey) + throws GroupDiscussionException { if (cancellation != null && cancellation.isCancelled()) { throw new GroupConversationService.MemberTurnCancelledException(); diff --git a/src/main/java/ai/labs/eddi/engine/lifecycle/GroupConversationEventSink.java b/src/main/java/ai/labs/eddi/engine/lifecycle/GroupConversationEventSink.java index 8182bcebd..aa08a2f25 100644 --- a/src/main/java/ai/labs/eddi/engine/lifecycle/GroupConversationEventSink.java +++ b/src/main/java/ai/labs/eddi/engine/lifecycle/GroupConversationEventSink.java @@ -61,6 +61,13 @@ private GroupConversationEventSink() { * Always preceded by an {@link #EVENT_CONVERGENCE_CHECKED} for the same repeat. */ public static final String EVENT_CONVERGENCE_REACHED = "convergence_reached"; + /** + * A member created or updated a shared artifact (I17). Fired by the turn + * executor after the turn that made the write โ€” tools have no listener + * reference, so accepted writes ride the live discussion's artifact-change + * queue until the executor drains it. + */ + public static final String EVENT_ARTIFACT_UPDATED = "artifact_updated"; // --- Event payloads --- @@ -159,4 +166,17 @@ public record ConvergenceCheckedEvent(int phaseIndex, String phaseName, int repe */ public record ConvergenceReachedEvent(int phaseIndex, String phaseName, int repeat, int repeatsSkipped, String reason) { } + + /** + * A member created or updated a shared artifact (I17). Carries metadata only, + * never the content โ€” an SSE observer reads the artifact through the REST + * payload, and content can be a quarter megabyte. + * + * @param created + * {@code true} for a fresh artifact (v1), {@code false} for an + * accepted update + */ + public record ArtifactUpdatedEvent(String artifactId, String name, String type, long version, String editorAgentId, + String status, boolean created) { + } } diff --git a/src/main/java/ai/labs/eddi/integrations/slack/SlackGroupDiscussionListener.java b/src/main/java/ai/labs/eddi/integrations/slack/SlackGroupDiscussionListener.java index b6e8cee99..5cc4b4876 100644 --- a/src/main/java/ai/labs/eddi/integrations/slack/SlackGroupDiscussionListener.java +++ b/src/main/java/ai/labs/eddi/integrations/slack/SlackGroupDiscussionListener.java @@ -295,6 +295,39 @@ public void onDecisionReached(GroupConversationEventSink.DecisionReachedEvent ev postSafe(channelId, threadTs, sb.toString().stripTrailing()); } + @Override + public void onArtifactUpdated(GroupConversationEventSink.ArtifactUpdatedEvent event) { + // Degenerate payload โ†’ skip rather than posting noise, same as + // onDecisionReached's NONE guard. + if (event == null || event.name() == null) { + return; + } + String emoji = event.created() ? "๐Ÿ“„" : "โœ๏ธ"; + String verb = event.created() ? "created" : "updated"; + var sb = new StringBuilder(); + sb.append(String.format("%s *Artifact \"%s\"* %s (v%d)", emoji, escapeMrkdwn(event.name()), verb, event.version())); + if (event.editorAgentId() != null && !event.editorAgentId().isBlank()) { + sb.append(String.format(" by %s", escapeMrkdwn(event.editorAgentId()))); + } + if ("FINAL".equals(event.status())) { + sb.append(" โ€” FINAL"); + } + + String threadTs = expandedMode ? null : userThreadTs; + postSafe(channelId, threadTs, sb.toString().stripTrailing()); + } + + /** + * Escapes Slack's three mrkdwn control characters. Without this, an + * LLM-authored artifact name (or an agent id) containing e.g. + * {@code } renders as a real channel broadcast. + */ + private static String escapeMrkdwn(String value) { + return value == null + ? null + : value.replace("&", "&").replace("<", "<").replace(">", ">"); + } + // โ”€โ”€โ”€ HITL (human-in-the-loop) โ”€โ”€โ”€ @Override diff --git a/src/main/java/ai/labs/eddi/modules/llm/impl/AgentOrchestrator.java b/src/main/java/ai/labs/eddi/modules/llm/impl/AgentOrchestrator.java index dd1248207..4bf556be1 100644 --- a/src/main/java/ai/labs/eddi/modules/llm/impl/AgentOrchestrator.java +++ b/src/main/java/ai/labs/eddi/modules/llm/impl/AgentOrchestrator.java @@ -49,6 +49,7 @@ import jakarta.enterprise.context.ApplicationScoped; import ai.labs.eddi.engine.internal.groups.LiveDiscussionRegistry; import ai.labs.eddi.configs.groups.IAgentGroupStore; +import ai.labs.eddi.configs.groups.ISharedArtifactStore; import jakarta.inject.Inject; import org.eclipse.microprofile.config.ConfigProvider; import org.jboss.logging.Logger; @@ -267,6 +268,13 @@ static void warnAboutUnenforcedBudgets(LlmConfiguration.Task task) { @Inject volatile IAgentGroupStore agentGroupStore; + /** + * I17's store, field-injected for the same reason as the two above. + * {@code ArtifactToolsProvider} treats null as "no artifact tools". + */ + @Inject + volatile ISharedArtifactStore sharedArtifactStore; + /** * Test seam for supplying the attachment services to a directly-constructed * orchestrator (CDI populates the fields above in production). Previously this @@ -644,6 +652,7 @@ ToolSetup buildToolSetup(LlmConfiguration.Task task, IConversationMemory memory) var contextual = contextualToolsProvider(); merger.addAll(List.of(builtinToolsProvider, contextual, dynamicAgentToolsProvider(), new GroupTaskToolsProvider(liveDiscussionRegistry, agentGroupStore), + new ArtifactToolsProvider(liveDiscussionRegistry, agentGroupStore, sharedArtifactStore), new AttachmentToolsProvider(contextual)), ctx); // LAZY registers every built-in's executor but shows the model only diff --git a/src/main/java/ai/labs/eddi/modules/llm/impl/ArtifactToolsProvider.java b/src/main/java/ai/labs/eddi/modules/llm/impl/ArtifactToolsProvider.java new file mode 100644 index 000000000..e7d78b3f0 --- /dev/null +++ b/src/main/java/ai/labs/eddi/modules/llm/impl/ArtifactToolsProvider.java @@ -0,0 +1,105 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.modules.llm.impl; + +import ai.labs.eddi.configs.groups.IAgentGroupStore; +import ai.labs.eddi.configs.groups.ISharedArtifactStore; +import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration; +import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.ArtifactConfig; +import ai.labs.eddi.engine.internal.groups.LiveDiscussionRegistry; +import ai.labs.eddi.modules.llm.tools.impl.ArtifactTools; +import ai.labs.eddi.modules.llm.tools.spi.ToolAssemblyContext; +import ai.labs.eddi.modules.llm.tools.spi.ToolContribution; +import ai.labs.eddi.modules.llm.tools.spi.ToolSourceProvider; +import org.jboss.logging.Logger; + +import java.util.List; +import java.util.Map; + +/** + * Contributes the shared-artifact tools (I17) โ€” {@code createArtifact}, + * {@code readArtifact}, {@code proposeArtifactUpdate}, {@code listArtifacts} โ€” + * when, and only when, the turn belongs to a live group discussion whose config + * sets {@code artifactConfig.allowArtifactTools}. + *

    + * Gate discipline is identical to {@link GroupTaskToolsProvider}, and for the + * same reasons: membership (not existence) via + * {@code LiveDiscussionRegistry#getForMember} because the group conversation id + * is a caller-supplied context variable; the member agent's own + * {@code enableBuiltInTools} switch still applies; and every uncertainty โ€” + * missing config, unreadable group, absent store โ€” resolves to "contribute + * nothing" (fail-closed, gate by absence). + * + * @author ginccc + */ +class ArtifactToolsProvider implements ToolSourceProvider { + + private static final Logger LOGGER = Logger.getLogger(ArtifactToolsProvider.class); + + private final LiveDiscussionRegistry liveDiscussionRegistry; + private final IAgentGroupStore groupStore; + private final ISharedArtifactStore artifactStore; + + ArtifactToolsProvider(LiveDiscussionRegistry liveDiscussionRegistry, IAgentGroupStore groupStore, + ISharedArtifactStore artifactStore) { + this.liveDiscussionRegistry = liveDiscussionRegistry; + this.groupStore = groupStore; + this.artifactStore = artifactStore; + } + + @Override + public String source() { + return "builtin"; + } + + @Override + public ToolContribution contribute(ToolAssemblyContext ctx) { + String groupConversationId = ctx.groupConversationId(); + if (groupConversationId == null || liveDiscussionRegistry == null || groupStore == null || artifactStore == null) { + return ToolContribution.empty(); + } + Boolean enableBuiltInTools = ctx.task() != null ? ctx.task().getEnableBuiltInTools() : null; + if (enableBuiltInTools == null || !enableBuiltInTools) { + return ToolContribution.empty(); + } + String callerConversationId = ctx.memory() != null ? ctx.memory().getConversationId() : null; + var live = liveDiscussionRegistry.getForMember(groupConversationId, callerConversationId); + if (live.isEmpty()) { + return ToolContribution.empty(); + } + AgentGroupConfiguration groupConfiguration = resolveGroup(live.get().getGroupId()); + ArtifactConfig config = groupConfiguration != null ? groupConfiguration.getArtifactConfig() : null; + if (config == null || !config.allowArtifactTools()) { + return ToolContribution.empty(); + } + + var tools = List.of(new ArtifactTools(liveDiscussionRegistry, groupConversationId, config, ctx.agentId(), + artifactStore)); + var reflected = ToolObjectReflector.reflect(tools); + return new ToolContribution(reflected.specs(), reflected.executors(), reflected.toolSources(), Map.of(), + List.of(), reflected.toolCanonicalNames()); + } + + /** + * The group config, or {@code null} if it cannot be read โ€” a store failure + * withholds the write tools (fail-closed), logged so the operator sees why they + * vanished. + */ + private AgentGroupConfiguration resolveGroup(String groupId) { + if (groupId == null) { + return null; + } + try { + var resourceId = groupStore.getCurrentResourceId(groupId); + if (resourceId == null) { + return null; + } + return groupStore.read(groupId, resourceId.getVersion()); + } catch (Exception e) { + LOGGER.warnf("Could not read artifact policy for group '%s' โ€” withholding the artifact tools: %s", groupId, e.getMessage()); + return null; + } + } +} diff --git a/src/main/java/ai/labs/eddi/modules/llm/tools/impl/ArtifactTools.java b/src/main/java/ai/labs/eddi/modules/llm/tools/impl/ArtifactTools.java new file mode 100644 index 000000000..fade27731 --- /dev/null +++ b/src/main/java/ai/labs/eddi/modules/llm/tools/impl/ArtifactTools.java @@ -0,0 +1,312 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.modules.llm.tools.impl; + +import ai.labs.eddi.configs.groups.ArtifactValidators; +import ai.labs.eddi.configs.groups.ISharedArtifactStore; +import ai.labs.eddi.configs.groups.ISharedArtifactStore.ArtifactGoneException; +import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.ArtifactConfig; +import ai.labs.eddi.configs.groups.model.GroupConversation; +import ai.labs.eddi.configs.groups.model.GroupConversation.ArtifactChange; +import ai.labs.eddi.configs.groups.model.SharedArtifact; +import ai.labs.eddi.configs.groups.model.SharedArtifact.ArtifactStatus; +import ai.labs.eddi.configs.groups.model.SharedArtifact.ArtifactType; +import ai.labs.eddi.datastore.IResourceStore; +import ai.labs.eddi.engine.internal.groups.LiveDiscussionRegistry; +import dev.langchain4j.agent.tool.P; +import dev.langchain4j.agent.tool.Tool; +import jakarta.enterprise.inject.Vetoed; +import org.jboss.logging.Logger; + +import java.time.Instant; +import java.util.List; +import java.util.Locale; + +/** + * Lets group members create and co-edit shared artifacts (I17, blackboard-lite) + * โ€” typed documents that live outside the dialogue, so structured work + * stops being prose the next agent re-parses. + *

    + * Writes go through the artifact store directly. Artifacts have their + * own collection, so the F1 rule that task-list writes must mutate the live + * {@link GroupConversation} instance (or be clobbered by the loop's next + * whole-document persist) does not apply. The live registry is still consulted + * โ€” a write against a finished or paused discussion is refused, and the + * creation/edit is announced through the live instance's artifact-change queue + * so the discussion loop can fire {@code artifact_updated}. + *

    + * Concurrency is deterministic CAS-and-retry, not an LLM merge. Every + * update presents the version it read; a stale version gets a "re-read and + * merge" sentence and the model retries against fresh content. Creation + * (duplicate-name + count cap) is a check-then-act, so it synchronizes on the + * live discussion instance โ€” PARALLEL phases genuinely run members at once. + *

    + * Constructed per-turn with runtime values โ€” NOT a CDI bean. {@code @Vetoed} is + * load-bearing for the same deployment-failure reason as + * {@link GroupTaskTools}. + * + * @author ginccc + */ +@Vetoed +public class ArtifactTools { + + private static final Logger LOGGER = Logger.getLogger(ArtifactTools.class); + + static final int MAX_NAME_LENGTH = 200; + + private final LiveDiscussionRegistry registry; + private final String groupConversationId; + private final ArtifactConfig config; + private final String agentId; + private final ISharedArtifactStore artifactStore; + + public ArtifactTools(LiveDiscussionRegistry registry, String groupConversationId, ArtifactConfig config, String agentId, + ISharedArtifactStore artifactStore) { + this.registry = registry; + this.groupConversationId = groupConversationId; + this.config = config; + this.agentId = agentId; + this.artifactStore = artifactStore; + } + + @Tool("Create a new shared artifact โ€” a named document the whole team can read and propose changes to. " + + "Use listArtifacts first to avoid duplicating one that already exists.") + public String createArtifact( + @P("Short unique name, how the team refers to this artifact") String name, + @P("Content type: TEXT, MARKDOWN or JSON") String type, + @P("The initial content") String content) { + + GroupConversation gc = liveDiscussion(); + if (gc == null) { + return "This discussion is no longer accepting artifact changes (it has finished or is paused)."; + } + if (name == null || name.isBlank()) { + return "An artifact needs a name."; + } + String trimmedName = name.trim(); + if (trimmedName.length() > MAX_NAME_LENGTH) { + return "The name is too long (%d chars, max %d). Put the detail in the content.".formatted(trimmedName.length(), MAX_NAME_LENGTH); + } + ArtifactType artifactType = parseType(type); + if (artifactType == null) { + return "Unknown artifact type \"%s\". Use TEXT, MARKDOWN or JSON.".formatted(type); + } + String rejection = checkContent(content); + if (rejection != null) { + return rejection; + } + + // Check-then-act (duplicate name + count cap) under one monitor: the live + // instance is the one object every member of this discussion shares in + // this JVM, so it is the natural lock for creation races. Updates need no + // lock โ€” the store's version CAS decides those. + synchronized (gc) { + List existing; + try { + existing = artifactStore.listByGroupConversationId(groupConversationId); + } catch (IResourceStore.ResourceStoreException e) { + LOGGER.warnf("Could not list artifacts for %s โ€” refusing the create: %s", groupConversationId, e.getMessage()); + return "The artifact store is unavailable right now; try again next turn."; + } + if (existing.stream().anyMatch(a -> trimmedName.equalsIgnoreCase(a.getName()))) { + return "An artifact named \"%s\" already exists. Read it with readArtifact and use proposeArtifactUpdate to change it." + .formatted(trimmedName); + } + if (existing.size() >= config.maxArtifactsPerDiscussion()) { + return "This discussion already has %d artifacts, which is the limit. Update an existing one instead of creating more." + .formatted(existing.size()); + } + + var artifact = new SharedArtifact(); + artifact.setGroupConversationId(groupConversationId); + artifact.setOwnerUserId(gc.getUserId()); + artifact.setName(trimmedName); + artifact.setType(artifactType); + artifact.setContent(content); + artifact.setVersion(1); + artifact.setLastEditorAgentId(agentId); + artifact.setStatus(ArtifactStatus.DRAFT); + artifact.setCreatedAt(Instant.now()); + artifact.setUpdatedAt(artifact.getCreatedAt()); + try { + artifactStore.create(artifact); + } catch (IResourceStore.ResourceStoreException e) { + LOGGER.warnf("Could not create artifact '%s' for %s: %s", trimmedName, groupConversationId, e.getMessage()); + return "The artifact store is unavailable right now; try again next turn."; + } + + gc.queueArtifactChange(new ArtifactChange(artifact.getId(), trimmedName, artifactType.name(), 1, agentId, + ArtifactStatus.DRAFT.name(), true)); + LOGGER.infof("Agent '%s' created artifact '%s' (v1) on group conversation %s", agentId, trimmedName, groupConversationId); + return "Created artifact \"%s\" (v1). The team can read it with readArtifact and change it with proposeArtifactUpdate." + .formatted(trimmedName); + } + } + + @Tool("Read a shared artifact's current content and version. You need the version to propose an update.") + public String readArtifact(@P("The artifact's name (or id)") String nameOrId) { + SharedArtifact artifact = resolve(nameOrId); + if (artifact == null) { + return "No artifact named \"%s\" here. Use listArtifacts to see what exists.".formatted(nameOrId != null ? nameOrId.trim() : ""); + } + return "Artifact \"%s\" (%s, %s, v%d, last edited by %s):\n%s".formatted( + artifact.getName(), artifact.getType(), artifact.getStatus(), artifact.getVersion(), + artifact.getLastEditorAgentId() != null ? artifact.getLastEditorAgentId() : "unknown", + artifact.getContent() != null ? artifact.getContent() : ""); + } + + @Tool("Propose new content for a shared artifact. You must pass the version you READ โ€” if someone changed the " + + "artifact since, your update is rejected and you re-read, merge your change into theirs, and retry.") + public String proposeArtifactUpdate( + @P("The artifact's name (or id)") String nameOrId, + @P("The complete new content (it replaces the old content)") String content, + @P("The version you read, from readArtifact") long expectedVersion, + @P(value = "Pass true to freeze the artifact as FINAL after this update. " + + "A FINAL artifact accepts no further updates.", + required = false) Boolean markFinal) { + + GroupConversation gc = liveDiscussion(); + if (gc == null) { + return "This discussion is no longer accepting artifact changes (it has finished or is paused)."; + } + SharedArtifact artifact = resolve(nameOrId); + if (artifact == null) { + return "No artifact named \"%s\" here. Use listArtifacts to see what exists.".formatted(nameOrId != null ? nameOrId.trim() : ""); + } + if (artifact.getStatus() == ArtifactStatus.FINAL) { + return "Artifact \"%s\" is FINAL and accepts no further updates.".formatted(artifact.getName()); + } + String rejection = checkContent(content); + if (rejection != null) { + return rejection; + } + if (artifact.getVersion() != expectedVersion) { + return staleVersion(artifact.getName(), artifact.getVersion()); + } + + artifact.applyEdit(content, agentId, Instant.now()); + if (Boolean.TRUE.equals(markFinal)) { + artifact.setStatus(ArtifactStatus.FINAL); + } + try { + artifactStore.updateIfVersion(artifact, expectedVersion); + } catch (IResourceStore.ResourceModifiedException e) { + // Lost the race after our read โ€” report the CURRENT version so the + // retry is usable. A failed re-read still yields the retry instruction. + long nowVersion = currentVersionOf(artifact.getId()); + return nowVersion > 0 + ? staleVersion(artifact.getName(), nowVersion) + : "Artifact \"%s\" changed since you read it; re-read and merge your change.".formatted(artifact.getName()); + } catch (ArtifactGoneException e) { + return "Artifact \"%s\" no longer exists.".formatted(artifact.getName()); + } catch (IResourceStore.ResourceStoreException e) { + LOGGER.warnf("Could not update artifact '%s' for %s: %s", artifact.getName(), groupConversationId, e.getMessage()); + return "The artifact store is unavailable right now; try again next turn."; + } + + gc.queueArtifactChange(new ArtifactChange(artifact.getId(), artifact.getName(), artifact.getType().name(), + artifact.getVersion(), agentId, artifact.getStatus().name(), false)); + LOGGER.infof("Agent '%s' updated artifact '%s' to v%d on group conversation %s", + agentId, artifact.getName(), artifact.getVersion(), groupConversationId); + return "Updated \"%s\" to v%d.%s".formatted(artifact.getName(), artifact.getVersion(), + artifact.getStatus() == ArtifactStatus.FINAL ? " It is now FINAL." : ""); + } + + @Tool("List this discussion's shared artifacts with their type, status and current version.") + public String listArtifacts() { + List artifacts; + try { + artifacts = artifactStore.listByGroupConversationId(groupConversationId); + } catch (IResourceStore.ResourceStoreException e) { + return "The artifact store is unavailable right now; try again next turn."; + } + if (artifacts.isEmpty()) { + return "No artifacts yet. Create one with createArtifact."; + } + var sb = new StringBuilder("Shared artifacts:\n"); + for (SharedArtifact a : artifacts) { + sb.append("- \"").append(a.getName()).append("\" (").append(a.getType()).append(", ").append(a.getStatus()) + .append(", v").append(a.getVersion()).append(')'); + if (a.getLastEditorAgentId() != null) { + sb.append(" โ€” last edited by ").append(a.getLastEditorAgentId()); + } + sb.append('\n'); + } + return sb.toString().stripTrailing(); + } + + /** The plan-specified stale-CAS sentence, verbatim shape. */ + private static String staleVersion(String name, long currentVersion) { + return "Artifact \"%s\" changed since you read it (now v%d); re-read and merge your change.".formatted(name, currentVersion); + } + + /** + * Resolves an artifact by name or id โ€” but only among THIS discussion's + * artifacts. Never a raw store read of a caller-supplied id: the id is + * model-controlled text, and resolving it globally would read another + * discussion's artifact. + */ + private SharedArtifact resolve(String nameOrId) { + if (nameOrId == null || nameOrId.isBlank()) { + return null; + } + String wanted = nameOrId.trim(); + List artifacts; + try { + artifacts = artifactStore.listByGroupConversationId(groupConversationId); + } catch (IResourceStore.ResourceStoreException e) { + LOGGER.warnf("Could not resolve artifact '%s' for %s: %s", wanted, groupConversationId, e.getMessage()); + return null; + } + return artifacts.stream() + .filter(a -> wanted.equalsIgnoreCase(a.getName()) || wanted.equals(a.getId())) + .findFirst().orElse(null); + } + + /** Size cap first (cheap), then the config's declarative validator chain. */ + private String checkContent(String content) { + if (content == null || content.isBlank()) { + return "An artifact needs content."; + } + int bytes = SharedArtifact.contentBytes(content); + if (bytes > SharedArtifact.MAX_CONTENT_BYTES) { + // Ceil, not truncate: MAX+1 bytes must not read "256 KB is over the + // 256 KB limit" โ€” the model retries at the same size on that sentence. + return "The content is %d KB, over the %d KB limit for a single artifact. Split it or shorten it." + .formatted(Math.ceilDiv(bytes, 1024), SharedArtifact.MAX_CONTENT_BYTES / 1024); + } + return ArtifactValidators.firstRejection(config.validators(), content); + } + + private static ArtifactType parseType(String type) { + if (type == null || type.isBlank()) { + return null; + } + try { + return ArtifactType.valueOf(type.trim().toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException e) { + return null; + } + } + + /** + * Live-instance lookup for writes. Authorization happened at assembly time (the + * provider resolves membership via {@code getForMember}); this lookup only + * answers "is the discussion still running", so a paused or finished discussion + * refuses instead of accepting a write nobody will announce. + */ + private GroupConversation liveDiscussion() { + return registry.get(groupConversationId).orElse(null); + } + + /** The stored artifact's current version, or 0 when unreadable. */ + private long currentVersionOf(String artifactId) { + try { + return artifactStore.read(artifactId).getVersion(); + } catch (Exception e) { + return 0; + } + } +} diff --git a/src/test/java/ai/labs/eddi/configs/groups/ArtifactValidatorsTest.java b/src/test/java/ai/labs/eddi/configs/groups/ArtifactValidatorsTest.java new file mode 100644 index 000000000..bf94e43ba --- /dev/null +++ b/src/test/java/ai/labs/eddi/configs/groups/ArtifactValidatorsTest.java @@ -0,0 +1,189 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.configs.groups; + +import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.ArtifactConfig; +import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.ArtifactValidator; +import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.ValidatorKind; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * I17 โ€” the declarative artifact validation chain: save-time spec checks fail + * the config save with an actionable path, write-time failures come back as + * rejection sentences for the model, and a broken spec fails closed. + * + * @author tests + */ +class ArtifactValidatorsTest { + + private static final String SCHEMA = """ + {"type":"object","required":["title"],"properties":{"title":{"type":"string"}}}"""; + + private static ArtifactConfig config(ArtifactValidator... validators) { + return new ArtifactConfig(true, 5, List.of(validators)); + } + + // ================================================================= + // save-time: requireValidSpecs + // ================================================================= + + @Test + @DisplayName("valid specs of all three kinds pass the save-time check") + void validSpecs_pass() { + assertDoesNotThrow(() -> ArtifactValidators.requireValidSpecs(config( + new ArtifactValidator(ValidatorKind.JSON_SCHEMA, SCHEMA), + new ArtifactValidator(ValidatorKind.REGEX, "^#"), + new ArtifactValidator(ValidatorKind.MAX_LENGTH, "1000")))); + } + + @Test + @DisplayName("null config or empty chain is a no-op") + void absentConfig_noOp() { + assertDoesNotThrow(() -> ArtifactValidators.requireValidSpecs(null)); + assertDoesNotThrow(() -> ArtifactValidators.requireValidSpecs(new ArtifactConfig(true, 5, null))); + } + + @Test + @DisplayName("a broken spec fails the save and names the validator's position") + void brokenSpecs_throwWithPath() { + var badRegex = assertThrows(IllegalArgumentException.class, + () -> ArtifactValidators.requireValidSpecs(config(new ArtifactValidator(ValidatorKind.REGEX, "[unclosed")))); + assertTrue(badRegex.getMessage().contains("validators[0]"), badRegex.getMessage()); + + var badLength = assertThrows(IllegalArgumentException.class, + () -> ArtifactValidators.requireValidSpecs(config( + new ArtifactValidator(ValidatorKind.REGEX, "ok"), + new ArtifactValidator(ValidatorKind.MAX_LENGTH, "lots")))); + assertTrue(badLength.getMessage().contains("validators[1]"), badLength.getMessage()); + + assertThrows(IllegalArgumentException.class, + () -> ArtifactValidators.requireValidSpecs(config(new ArtifactValidator(ValidatorKind.MAX_LENGTH, "0")))); + assertThrows(IllegalArgumentException.class, + () -> ArtifactValidators.requireValidSpecs(config(new ArtifactValidator(null, "x")))); + assertThrows(IllegalArgumentException.class, + () -> ArtifactValidators.requireValidSpecs(config(new ArtifactValidator(ValidatorKind.JSON_SCHEMA, " ")))); + } + + @Test + @DisplayName("a schema with an invalid keyword VALUE fails the save โ€” parse alone would admit it") + void invalidKeywordValue_failsMetaSchema() { + // {"type":"strng"} is perfectly valid JSON and getSchema() parses it; + // only meta-schema validation catches the typo'd simple type. + var e = assertThrows(IllegalArgumentException.class, + () -> ArtifactValidators.requireValidSpecs(config( + new ArtifactValidator(ValidatorKind.JSON_SCHEMA, "{\"type\":\"strng\"}")))); + assertTrue(e.getMessage().contains("validators[0]"), e.getMessage()); + } + + @Test + @DisplayName("a null validator ENTRY fails the save with the positional message, not an NPE") + void nullValidatorEntry_actionableNotNpe() { + // Arrays.asList permits the null element; the config copy must too + // (List.copyOf would NPE during deserialization, preempting this message). + var config = new ArtifactConfig(true, 5, Arrays.asList( + new ArtifactValidator(ValidatorKind.MAX_LENGTH, "10"), null)); + + var e = assertThrows(IllegalArgumentException.class, () -> ArtifactValidators.requireValidSpecs(config)); + assertTrue(e.getMessage().contains("validators[1]"), e.getMessage()); + } + + // ================================================================= + // write-time: firstRejection + // ================================================================= + + @Test + @DisplayName("content passing the whole chain yields null") + void passingContent_null() { + var validators = List.of( + new ArtifactValidator(ValidatorKind.MAX_LENGTH, "100"), + new ArtifactValidator(ValidatorKind.REGEX, "title"), + new ArtifactValidator(ValidatorKind.JSON_SCHEMA, SCHEMA)); + + assertNull(ArtifactValidators.firstRejection(validators, "{\"title\":\"ok\"}")); + } + + @Test + @DisplayName("MAX_LENGTH rejection names both counts so the model can act") + void maxLength_rejects() { + String rejection = ArtifactValidators.firstRejection( + List.of(new ArtifactValidator(ValidatorKind.MAX_LENGTH, "5")), "123456"); + + assertNotNull(rejection); + assertTrue(rejection.contains("6") && rejection.contains("5"), rejection); + } + + @Test + @DisplayName("REGEX requires a match somewhere in the content") + void regex_rejectsAndPasses() { + var validators = List.of(new ArtifactValidator(ValidatorKind.REGEX, "^# .+")); + + assertNull(ArtifactValidators.firstRejection(validators, "# Heading\nbody")); + String rejection = ArtifactValidators.firstRejection(validators, "no heading here"); + assertNotNull(rejection); + assertTrue(rejection.contains("pattern"), rejection); + } + + @Test + @DisplayName("JSON_SCHEMA distinguishes 'not JSON' from 'JSON that violates the schema'") + void jsonSchema_rejections() { + var validators = List.of(new ArtifactValidator(ValidatorKind.JSON_SCHEMA, SCHEMA)); + + String notJson = ArtifactValidators.firstRejection(validators, "plain prose"); + assertNotNull(notJson); + assertTrue(notJson.contains("valid JSON"), notJson); + + String violates = ArtifactValidators.firstRejection(validators, "{\"other\":1}"); + assertNotNull(violates); + assertTrue(violates.contains("title"), "the violation message must name what is missing: " + violates); + + assertNull(ArtifactValidators.firstRejection(validators, "{\"title\":\"x\"}")); + } + + @Test + @DisplayName("the chain runs in config order โ€” the first failure wins") + void chainOrder_firstFailureWins() { + var validators = List.of( + new ArtifactValidator(ValidatorKind.MAX_LENGTH, "3"), + new ArtifactValidator(ValidatorKind.REGEX, "nope")); + + String rejection = ArtifactValidators.firstRejection(validators, "12345"); + assertNotNull(rejection); + assertTrue(rejection.contains("character"), "the length failure comes first: " + rejection); + } + + @Test + @DisplayName("a catastrophically backtracking pattern aborts on its deadline instead of pinning the turn") + void catastrophicRegex_abortsOnDeadline() { + // (a+)+$ against a long non-matching tail backtracks exponentially โ€” + // unguarded it runs for astronomical time, not the ~500ms budget. + var validators = List.of(new ArtifactValidator(ValidatorKind.REGEX, "(a+)+$")); + String content = "a".repeat(60_000) + "b"; + + long start = System.nanoTime(); + String rejection = ArtifactValidators.firstRejection(validators, content); + long elapsedMs = (System.nanoTime() - start) / 1_000_000; + + assertNotNull(rejection, "a runaway match must refuse the write, not admit it"); + assertTrue(rejection.contains("did not finish in time"), rejection); + assertTrue(elapsedMs < 5_000, "the deadline guard must abort far below the member-turn timeout, took " + elapsedMs + "ms"); + } + + @Test + @DisplayName("a broken spec at write time fails closed โ€” the write is refused, never admitted") + void brokenSpecAtWriteTime_failsClosed() { + assertNotNull(ArtifactValidators.firstRejection( + List.of(new ArtifactValidator(ValidatorKind.REGEX, "[unclosed")), "anything")); + assertNotNull(ArtifactValidators.firstRejection( + List.of(new ArtifactValidator(ValidatorKind.MAX_LENGTH, "many")), "anything")); + assertNotNull(ArtifactValidators.firstRejection( + List.of(new ArtifactValidator(null, null)), "anything")); + } +} diff --git a/src/test/java/ai/labs/eddi/configs/groups/mongo/SharedArtifactStoreTest.java b/src/test/java/ai/labs/eddi/configs/groups/mongo/SharedArtifactStoreTest.java new file mode 100644 index 000000000..7be2198ec --- /dev/null +++ b/src/test/java/ai/labs/eddi/configs/groups/mongo/SharedArtifactStoreTest.java @@ -0,0 +1,292 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.configs.groups.mongo; + +import ai.labs.eddi.configs.groups.ISharedArtifactStore.ArtifactGoneException; +import ai.labs.eddi.configs.groups.model.SharedArtifact; +import ai.labs.eddi.datastore.IResourceFilter; +import ai.labs.eddi.datastore.IResourceStorage; +import ai.labs.eddi.datastore.IResourceStorageFactory; +import ai.labs.eddi.datastore.IResourceStore; +import ai.labs.eddi.datastore.serialization.IDocumentBuilder; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.io.IOException; +import java.time.Instant; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +/** + * I17 โ€” {@link SharedArtifactStore}: the version CAS goes through the numeric + * {@code storeIfFieldEquals} overload (never an unconditional store), filters + * are anchored and re-checked in Java, and the GDPR erasure sweep follows the + * group-conversation store's page/re-check/fail-loud contract. + * + * @author tests + */ +@SuppressWarnings("unchecked") +class SharedArtifactStoreTest { + + private IResourceStorage storage; + private SharedArtifactStore store; + + @BeforeEach + void setUp() { + IResourceStorageFactory storageFactory = mock(IResourceStorageFactory.class); + IDocumentBuilder documentBuilder = mock(IDocumentBuilder.class); + storage = mock(IResourceStorage.class); + // Must mirror the production create() call EXACTLY, index-hint varargs + // included โ€” a mismatch silently leaves the internal storage null. + when(storageFactory.create(eq("sharedartifacts"), eq(documentBuilder), eq(SharedArtifact.class), + eq("groupConversationId"), eq("ownerUserId"))).thenReturn(storage); + store = new SharedArtifactStore(storageFactory, documentBuilder); + } + + private static SharedArtifact artifact(String id, String gcId, String owner, long version) { + var a = new SharedArtifact(); + a.setId(id); + a.setGroupConversationId(gcId); + a.setOwnerUserId(owner); + a.setName("draft"); + a.setVersion(version); + return a; + } + + private IResourceStorage.IResource resource(String id, SharedArtifact data) throws IOException { + IResourceStorage.IResource resource = mock(IResourceStorage.IResource.class); + when(resource.getId()).thenReturn(id); + when(resource.getData()).thenReturn(data); + return resource; + } + + private IResourceStore.IResourceId resourceId(String id) { + return new IResourceStore.IResourceId() { + @Override + public String getId() { + return id; + } + + @Override + public Integer getVersion() { + return 1; + } + }; + } + + // ================================================================= + // CAS + // ================================================================= + + @Test + @DisplayName("updateIfVersion goes through the NUMERIC CAS overload, never an unconditional store") + void updateIfVersion_usesNumericCas() throws Exception { + var a = artifact("a-1", "gc-1", "user-1", 3); + IResourceStorage.IResource writeResource = resource("a-1", a); + when(storage.newResource("a-1", 1, a)).thenReturn(writeResource); + + store.updateIfVersion(a, 2); + + verify(storage).storeIfFieldEquals(writeResource, "version", 2L); + verify(storage, never()).store(any(IResourceStorage.IResource.class)); + verify(storage, never()).storeIfFieldEquals(any(), anyString(), anyString()); + } + + @Test + @DisplayName("a CAS conflict propagates as ResourceModifiedException (retry), a gone document as ArtifactGoneException") + void updateIfVersion_distinguishesConflictFromGone() throws Exception { + var a = artifact("a-1", "gc-1", "user-1", 3); + IResourceStorage.IResource writeResource = resource("a-1", a); + when(storage.newResource("a-1", 1, a)).thenReturn(writeResource); + + doThrow(new IResourceStore.ResourceModifiedException("conflict")) + .when(storage).storeIfFieldEquals(writeResource, "version", 2L); + assertThrows(IResourceStore.ResourceModifiedException.class, () -> store.updateIfVersion(a, 2)); + + doThrow(new IResourceStore.ResourceNotFoundException("gone")) + .when(storage).storeIfFieldEquals(writeResource, "version", 2L); + assertThrows(ArtifactGoneException.class, () -> store.updateIfVersion(a, 2)); + } + + // ================================================================= + // read / create + // ================================================================= + + @Test + @DisplayName("read maps a missing document to a not-found whose message does NOT embed the caller-supplied id") + void readNotFound_messageDoesNotEmbedTheId() { + when(storage.read("attacker