From 24e99abce8a68cb703805885462d679d00ccbe71 Mon Sep 17 00:00:00 2001 From: tsuz <6927131+tsuz@users.noreply.github.com> Date: Wed, 12 Aug 2026 22:21:23 +0900 Subject: [PATCH 1/3] feat: per-user system prompt via compacted user-settings topic Gated by ENABLE_USER_SETTING (default false). When enabled, the streams app creates the compacted {AGENT_NAME}-user-settings topic (keyed by user_id), materializes it as a KTable, and left-joins it into FullSessionContext during enrichment. The think consumer uses the joined system_prompt as the base prompt in place of the default (SYSTEM_PROMPT_FILE / built-in); memoir context is still appended on top. A tombstone reverts the user to the default prompt. The enrichment processor now re-keys to user_id once when either memoir or user settings is enabled, so enabling both costs a single repartition, and the settings join works with MEMOIR_ENABLED=false. Co-Authored-By: Claude Fable 5 --- .../streams/FlightDeckStreamsApp.java | 38 +++- .../io/flightdeck/streams/config/Topics.java | 9 + .../streams/model/FullSessionContext.java | 1 + .../streams/model/UserSettings.java | 15 ++ .../EnrichInputMessageProcessor.java | 82 +++++-- .../streams/UserSettingTopologyTest.java | 204 ++++++++++++++++++ .../EnrichInputMessageProcessorTest.java | 71 +++++- .../think/consumer/ThinkConsumer.java | 16 +- .../think/model/FullSessionContext.java | 10 +- .../think/consumer/SystemPromptTest.java | 30 +++ 10 files changed, 451 insertions(+), 25 deletions(-) create mode 100644 processor-apps/processing/src/main/java/io/flightdeck/streams/model/UserSettings.java create mode 100644 processor-apps/processing/src/test/java/io/flightdeck/streams/UserSettingTopologyTest.java diff --git a/processor-apps/processing/src/main/java/io/flightdeck/streams/FlightDeckStreamsApp.java b/processor-apps/processing/src/main/java/io/flightdeck/streams/FlightDeckStreamsApp.java index e83a22a..717a300 100644 --- a/processor-apps/processing/src/main/java/io/flightdeck/streams/FlightDeckStreamsApp.java +++ b/processor-apps/processing/src/main/java/io/flightdeck/streams/FlightDeckStreamsApp.java @@ -9,6 +9,7 @@ import io.flightdeck.streams.processors.SessionEndProcessor; import io.flightdeck.streams.processors.TransformToolUseDoneProcessor; import io.flightdeck.streams.model.ThinkResponse; +import io.flightdeck.streams.model.UserSettings; import io.flightdeck.streams.serdes.JsonSerde; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.AdminClientConfig; @@ -44,9 +45,14 @@ public class FlightDeckStreamsApp { static final boolean MEMOIR_ENABLED = Boolean.parseBoolean( System.getenv().getOrDefault("MEMOIR_ENABLED", "true")); + /** When true, the user-settings topic is joined into the enriched context. */ + static final boolean USER_SETTING_ENABLED = Boolean.parseBoolean( + System.getenv().getOrDefault("ENABLE_USER_SETTING", "false")); + static final String MEMOIR_CONTEXT_STORE = "memoir-context-store"; static final String THINK_RESPONSE_STORE = "think-response-store"; static final String REPLY_TO_STORE = "reply-to-store"; + static final String USER_SETTINGS_STORE = "user-settings-store"; /** Time-based expiry for reply-to routing state (default 24h). */ static final long REPLY_TO_STATE_TTL_MS = Long.parseLong( @@ -90,10 +96,14 @@ public static void main(String[] args) { /** Build and return the full topology (also used by tests). */ public static Topology buildTopology() { - return buildTopology(MEMOIR_ENABLED); + return buildTopology(MEMOIR_ENABLED, USER_SETTING_ENABLED); } static Topology buildTopology(boolean memoirEnabled) { + return buildTopology(memoirEnabled, false); + } + + static Topology buildTopology(boolean memoirEnabled, boolean userSettingEnabled) { StreamsBuilder builder = new StreamsBuilder(); // ── Shared KTable: memoir-context (only when memoir is enabled) ─────── @@ -109,6 +119,19 @@ static Topology buildTopology(boolean memoirEnabled) { ); } + // ── KTable: user-settings (only when ENABLE_USER_SETTING=true) ─────── + KTable userSettingsTable = null; + if (userSettingEnabled) { + userSettingsTable = builder.table( + Topics.USER_SETTINGS, + Consumed.with(Serdes.String(), JsonSerde.of(UserSettings.class)), + Materialized.as( + Stores.persistentKeyValueStore(USER_SETTINGS_STORE)) + .withKeySerde(Serdes.String()) + .withValueSerde(JsonSerde.of(UserSettings.class)) + ); + } + // ── Shared KStream + KTable: think-request-response ───────────────── // ONE source registration — shared across all processors that read // this topic, avoiding TopologyException conflicts. @@ -137,7 +160,7 @@ static Topology buildTopology(boolean memoirEnabled) { ); // ── Register each processor fragment ────────────────────────────────── - EnrichInputMessageProcessor.register(builder, memoirTable, thinkTable); + EnrichInputMessageProcessor.register(builder, memoirTable, thinkTable, userSettingsTable); ExtractToolUseItemsProcessor.register(builder, thinkStream); EndTurnProcessor.register(builder, thinkStream, replyToTable); AggregateToolExecutionResultProcessor.register(builder, thinkStream); @@ -152,6 +175,8 @@ static Topology buildTopology(boolean memoirEnabled) { log.info("Memoir is DISABLED"); } + log.info("User settings are {}", userSettingEnabled ? "ENABLED" : "DISABLED"); + return builder.build(); } @@ -188,6 +213,10 @@ private static void ensureTopicsExist(Properties streamsProps) { )); } + if (USER_SETTING_ENABLED) { + requiredTopics.add(Topics.USER_SETTINGS); + } + try (AdminClient admin = AdminClient.create(adminProps)) { Set existing = admin.listTopics().names().get(30, TimeUnit.SECONDS); @@ -203,6 +232,11 @@ private static void ensureTopicsExist(Properties streamsProps) { "retention.ms", String.valueOf(REPLY_TO_STATE_TTL_MS) )); } + // user-settings is pure keyed state: keep the latest + // settings record per user_id forever. + if (t.equals(Topics.USER_SETTINGS)) { + topic.configs(Map.of("cleanup.policy", "compact")); + } return topic; }) .collect(Collectors.toList()); diff --git a/processor-apps/processing/src/main/java/io/flightdeck/streams/config/Topics.java b/processor-apps/processing/src/main/java/io/flightdeck/streams/config/Topics.java index e645a2a..bfd914c 100644 --- a/processor-apps/processing/src/main/java/io/flightdeck/streams/config/Topics.java +++ b/processor-apps/processing/src/main/java/io/flightdeck/streams/config/Topics.java @@ -56,6 +56,15 @@ private static String requireEnv(String key) { /** Emitted when a session has been inactive for a configured period */ public static final String SESSION_END = PREFIX + "session-end"; + // ── User settings ───────────────────────────────────────────────────────── + /** + * Per-user settings (system prompt override, etc.), keyed by user_id. + * Compacted so the latest settings record per user is retained. Only wired + * into the topology when {@code ENABLE_USER_SETTING=true}; a tombstone + * (null value) reverts the user to the default settings. + */ + public static final String USER_SETTINGS = PREFIX + "user-settings"; + // ── Memoir ─────────────────────────────────────────────────────────────── /** Long-term memoir / summary context per session (KTable) */ public static final String MEMOIR_CONTEXT = PREFIX + "memoir-context"; diff --git a/processor-apps/processing/src/main/java/io/flightdeck/streams/model/FullSessionContext.java b/processor-apps/processing/src/main/java/io/flightdeck/streams/model/FullSessionContext.java index 714f6db..3083e7f 100644 --- a/processor-apps/processing/src/main/java/io/flightdeck/streams/model/FullSessionContext.java +++ b/processor-apps/processing/src/main/java/io/flightdeck/streams/model/FullSessionContext.java @@ -19,5 +19,6 @@ public record FullSessionContext( @JsonProperty("history") List history, @JsonProperty("latest_input") MessageInput latestInput, @JsonProperty("memoir_context") String memoirContext, + @JsonProperty("system_prompt") String systemPrompt, @JsonProperty("timestamp") String timestamp ) {} \ No newline at end of file diff --git a/processor-apps/processing/src/main/java/io/flightdeck/streams/model/UserSettings.java b/processor-apps/processing/src/main/java/io/flightdeck/streams/model/UserSettings.java new file mode 100644 index 0000000..f217b58 --- /dev/null +++ b/processor-apps/processing/src/main/java/io/flightdeck/streams/model/UserSettings.java @@ -0,0 +1,15 @@ +package io.flightdeck.streams.model; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** + * Per-user settings stored in the compacted {@code user-settings} topic, + * keyed by user_id. Joined into {@link FullSessionContext} by the + * {@code EnrichInputMessageProcessor} when {@code ENABLE_USER_SETTING=true}. + */ +@JsonIgnoreProperties(ignoreUnknown = true) +public record UserSettings( + @JsonProperty("system_prompt") String systemPrompt, + @JsonProperty("updated_at") String updatedAt +) {} diff --git a/processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EnrichInputMessageProcessor.java b/processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EnrichInputMessageProcessor.java index 7247c7a..7b0915c 100644 --- a/processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EnrichInputMessageProcessor.java +++ b/processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EnrichInputMessageProcessor.java @@ -4,6 +4,7 @@ import io.flightdeck.streams.model.MessageInput; import io.flightdeck.streams.model.FullSessionContext; import io.flightdeck.streams.model.ThinkResponse; +import io.flightdeck.streams.model.UserSettings; import io.flightdeck.streams.serdes.JsonSerde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsBuilder; @@ -28,8 +29,11 @@ * │ leftJoin * │◄────────────── memoir-context (KTable — long-term memoir, shared) * │ + * │ leftJoin + * │◄────────────── user-settings (KTable — per-user settings, optional) + * │ * ▼ - * enriched-message-input (KStream — history + memoir + latest input) + * enriched-message-input (KStream — history + memoir + settings + latest input) * * *

History is reconstructed from the previous ThinkResponse: @@ -40,12 +44,14 @@ public class EnrichInputMessageProcessor { private static final Logger log = LoggerFactory.getLogger(EnrichInputMessageProcessor.class); /** - * @param memoirTable shared KTable for memoir-context (keyed by userId) - * @param thinkTable shared KTable for think-request-response (keyed by sessionId) + * @param memoirTable shared KTable for memoir-context (keyed by userId), null when memoir is disabled + * @param thinkTable shared KTable for think-request-response (keyed by sessionId) + * @param userSettingsTable KTable for user-settings (keyed by userId), null when ENABLE_USER_SETTING=false */ public static void register(StreamsBuilder builder, KTable memoirTable, - KTable thinkTable) { + KTable thinkTable, + KTable userSettingsTable) { // ── Left side: incoming user messages ──────────────────────────────── KStream inputStream = builder.stream( @@ -69,29 +75,47 @@ public static void register(StreamsBuilder builder, ) ); - // If memoir is enabled, re-key by userId, join with memoir, re-key back - if (memoirTable != null) { - enriched = enriched + // If memoir or user settings are enabled, re-key by userId once, apply + // whichever user-keyed joins are active, then re-key back to sessionId. + if (memoirTable != null || userSettingsTable != null) { + KStream byUser = enriched .selectKey((sessionId, full) -> - full.userId() != null ? full.userId() : sessionId) - .leftJoin( - memoirTable, - EnrichInputMessageProcessor::enrichWithMemoir, - Joined.with( - Serdes.String(), - JsonSerde.of(FullSessionContext.class), - Serdes.String() - ) - ) - .selectKey((userId, full) -> full.sessionId()); + full.userId() != null ? full.userId() : sessionId); + + if (memoirTable != null) { + byUser = byUser.leftJoin( + memoirTable, + EnrichInputMessageProcessor::enrichWithMemoir, + Joined.with( + Serdes.String(), + JsonSerde.of(FullSessionContext.class), + Serdes.String() + ) + ); + } + + if (userSettingsTable != null) { + byUser = byUser.leftJoin( + userSettingsTable, + EnrichInputMessageProcessor::enrichWithUserSettings, + Joined.with( + Serdes.String(), + JsonSerde.of(FullSessionContext.class), + JsonSerde.of(UserSettings.class) + ) + ); + } + + enriched = byUser.selectKey((userId, full) -> full.sessionId()); } enriched .peek((sessionId, full) -> - log.info("[{}] Enriched — history_size={} has_memoir={} first_turn={}", + log.info("[{}] Enriched — history_size={} has_memoir={} has_system_prompt={} first_turn={}", sessionId, full.history().size(), full.memoirContext() != null, + full.systemPrompt() != null, full.history().isEmpty())) .to(Topics.ENRICHED_MESSAGE_INPUT, Produced.with(Serdes.String(), JsonSerde.of(FullSessionContext.class))); @@ -130,6 +154,7 @@ static FullSessionContext enrichWithThinkResponse(MessageInput message, ThinkRes history, message, null, + null, Instant.now().toString() ); } @@ -145,6 +170,25 @@ static FullSessionContext enrichWithMemoir(FullSessionContext enriched, String m enriched.history(), enriched.latestInput(), memoir, + enriched.systemPrompt(), + enriched.timestamp() + ); + } + + /** + * Join: attach the per-user system prompt override to the session context. + * A missing settings record (or one without a system_prompt) leaves the + * field null, which the think consumer treats as "use the default prompt". + */ + static FullSessionContext enrichWithUserSettings(FullSessionContext enriched, UserSettings settings) { + return new FullSessionContext( + enriched.sessionId(), + enriched.userId(), + enriched.cost(), + enriched.history(), + enriched.latestInput(), + enriched.memoirContext(), + settings != null ? settings.systemPrompt() : null, enriched.timestamp() ); } diff --git a/processor-apps/processing/src/test/java/io/flightdeck/streams/UserSettingTopologyTest.java b/processor-apps/processing/src/test/java/io/flightdeck/streams/UserSettingTopologyTest.java new file mode 100644 index 0000000..15f1a97 --- /dev/null +++ b/processor-apps/processing/src/test/java/io/flightdeck/streams/UserSettingTopologyTest.java @@ -0,0 +1,204 @@ +package io.flightdeck.streams; + +import io.flightdeck.streams.config.Topics; +import io.flightdeck.streams.model.FullSessionContext; +import io.flightdeck.streams.model.MessageInput; +import io.flightdeck.streams.model.UserSettings; +import io.flightdeck.streams.serdes.JsonSerde; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.*; +import org.junit.jupiter.api.*; + +import java.util.Map; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.*; + +/** + * Tests that the ENABLE_USER_SETTING flag correctly includes or excludes + * the user-settings KTable and its join from the topology. + */ +class UserSettingTopologyTest { + + private static final String TS = "2026-03-10T12:00:00Z"; + + // ── User settings ENABLED ─────────────────────────────────────────────── + + @Nested + @DisplayName("When ENABLE_USER_SETTING=true") + class UserSettingEnabled { + + private TopologyTestDriver driver; + private TestInputTopic messageInput; + private TestInputTopic userSettingsInput; + private TestOutputTopic enrichedOutput; + + @BeforeEach + void setUp() { + Topology topology = FlightDeckStreamsApp.buildTopology(true, true); + driver = new TopologyTestDriver(topology, testProps()); + + messageInput = driver.createInputTopic( + Topics.MESSAGE_INPUT, + Serdes.String().serializer(), + JsonSerde.of(MessageInput.class).serializer()); + + userSettingsInput = driver.createInputTopic( + Topics.USER_SETTINGS, + Serdes.String().serializer(), + JsonSerde.of(UserSettings.class).serializer()); + + enrichedOutput = driver.createOutputTopic( + Topics.ENRICHED_MESSAGE_INPUT, + Serdes.String().deserializer(), + JsonSerde.of(FullSessionContext.class).deserializer()); + } + + @AfterEach + void tearDown() { driver.close(); } + + @Test + @DisplayName("Per-user system prompt is joined into the enriched message") + void systemPromptJoinedIntoEnrichedMessage() { + userSettingsInput.pipeInput("user-1", new UserSettings("You are a billing agent.", TS)); + + messageInput.pipeInput("sess-1", userMsg("sess-1", "user-1", "Hello")); + + FullSessionContext result = enrichedOutput.readRecord().value(); + assertThat(result.systemPrompt()).isEqualTo("You are a billing agent."); + } + + @Test + @DisplayName("system_prompt stays null for users without a settings record") + void systemPromptNullWithoutSettings() { + messageInput.pipeInput("sess-2", userMsg("sess-2", "user-none", "Hello")); + + assertThat(enrichedOutput.readRecord().value().systemPrompt()).isNull(); + } + + @Test + @DisplayName("user-settings-store is registered in topology") + void userSettingsStoreRegistered() { + String description = FlightDeckStreamsApp.buildTopology(true, true) + .describe().toString(); + assertThat(description).contains(FlightDeckStreamsApp.USER_SETTINGS_STORE); + } + } + + // ── User settings ENABLED while memoir is DISABLED ────────────────────── + + @Nested + @DisplayName("When ENABLE_USER_SETTING=true and MEMOIR_ENABLED=false") + class UserSettingEnabledMemoirDisabled { + + private TopologyTestDriver driver; + private TestInputTopic messageInput; + private TestInputTopic userSettingsInput; + private TestOutputTopic enrichedOutput; + + @BeforeEach + void setUp() { + Topology topology = FlightDeckStreamsApp.buildTopology(false, true); + driver = new TopologyTestDriver(topology, testProps()); + + messageInput = driver.createInputTopic( + Topics.MESSAGE_INPUT, + Serdes.String().serializer(), + JsonSerde.of(MessageInput.class).serializer()); + + userSettingsInput = driver.createInputTopic( + Topics.USER_SETTINGS, + Serdes.String().serializer(), + JsonSerde.of(UserSettings.class).serializer()); + + enrichedOutput = driver.createOutputTopic( + Topics.ENRICHED_MESSAGE_INPUT, + Serdes.String().deserializer(), + JsonSerde.of(FullSessionContext.class).deserializer()); + } + + @AfterEach + void tearDown() { driver.close(); } + + @Test + @DisplayName("System prompt join works without the memoir join") + void systemPromptJoinWorksWithoutMemoir() { + userSettingsInput.pipeInput("user-3", new UserSettings("Custom prompt.", TS)); + + messageInput.pipeInput("sess-3", userMsg("sess-3", "user-3", "Hello")); + + FullSessionContext result = enrichedOutput.readRecord().value(); + assertThat(result.systemPrompt()).isEqualTo("Custom prompt."); + assertThat(result.memoirContext()).isNull(); + } + + @Test + @DisplayName("Output key is preserved as session_id after the user-keyed join") + void outputKeyPreserved() { + userSettingsInput.pipeInput("user-4", new UserSettings("p", TS)); + messageInput.pipeInput("sess-4", userMsg("sess-4", "user-4", "hi")); + + assertThat(enrichedOutput.readRecord().key()).isEqualTo("sess-4"); + } + } + + // ── User settings DISABLED ────────────────────────────────────────────── + + @Nested + @DisplayName("When ENABLE_USER_SETTING=false") + class UserSettingDisabled { + + private TopologyTestDriver driver; + private TestInputTopic messageInput; + private TestOutputTopic enrichedOutput; + + @BeforeEach + void setUp() { + Topology topology = FlightDeckStreamsApp.buildTopology(true, false); + driver = new TopologyTestDriver(topology, testProps()); + + messageInput = driver.createInputTopic( + Topics.MESSAGE_INPUT, + Serdes.String().serializer(), + JsonSerde.of(MessageInput.class).serializer()); + + enrichedOutput = driver.createOutputTopic( + Topics.ENRICHED_MESSAGE_INPUT, + Serdes.String().deserializer(), + JsonSerde.of(FullSessionContext.class).deserializer()); + } + + @AfterEach + void tearDown() { driver.close(); } + + @Test + @DisplayName("Enriched message has null system_prompt") + void systemPromptIsNull() { + messageInput.pipeInput("sess-5", userMsg("sess-5", "user-5", "Hello")); + + assertThat(enrichedOutput.readRecord().value().systemPrompt()).isNull(); + } + + @Test + @DisplayName("user-settings topic and store are not registered in topology") + void userSettingsNotInTopology() { + String description = FlightDeckStreamsApp.buildTopology(true, false) + .describe().toString(); + assertThat(description).doesNotContain(FlightDeckStreamsApp.USER_SETTINGS_STORE); + assertThat(description).doesNotContain(Topics.USER_SETTINGS); + } + } + + // ── Helpers ───────────────────────────────────────────────────────────── + + private static Properties testProps() { + Properties props = new Properties(); + props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-user-setting"); + props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092"); + return props; + } + + private static MessageInput userMsg(String sessionId, String userId, String content) { + return new MessageInput(sessionId, userId, "user", content, TS, Map.of()); + } +} diff --git a/processor-apps/processing/src/test/java/io/flightdeck/streams/processors/EnrichInputMessageProcessorTest.java b/processor-apps/processing/src/test/java/io/flightdeck/streams/processors/EnrichInputMessageProcessorTest.java index 526ccc6..0c4025f 100644 --- a/processor-apps/processing/src/test/java/io/flightdeck/streams/processors/EnrichInputMessageProcessorTest.java +++ b/processor-apps/processing/src/test/java/io/flightdeck/streams/processors/EnrichInputMessageProcessorTest.java @@ -4,6 +4,7 @@ import io.flightdeck.streams.model.FullSessionContext; import io.flightdeck.streams.model.MessageInput; import io.flightdeck.streams.model.ThinkResponse; +import io.flightdeck.streams.model.UserSettings; import io.flightdeck.streams.serdes.JsonSerde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.*; @@ -44,6 +45,9 @@ class EnrichInputMessageProcessorTest { /** Seeds the memoir-context KTable */ private TestInputTopic memoirInput; + /** Seeds the user-settings KTable */ + private TestInputTopic userSettingsInput; + @BeforeEach void setUp() { StreamsBuilder builder = new StreamsBuilder(); @@ -55,7 +59,10 @@ void setUp() { KTable thinkTable = builder.table( Topics.THINK_REQUEST_RESPONSE, Consumed.with(Serdes.String(), JsonSerde.of(ThinkResponse.class))); - EnrichInputMessageProcessor.register(builder, memoirTable, thinkTable); + KTable userSettingsTable = builder.table( + Topics.USER_SETTINGS, + Consumed.with(Serdes.String(), JsonSerde.of(UserSettings.class))); + EnrichInputMessageProcessor.register(builder, memoirTable, thinkTable, userSettingsTable); Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-enrich"); @@ -79,6 +86,11 @@ void setUp() { Serdes.String().serializer(), Serdes.String().serializer()); + userSettingsInput = driver.createInputTopic( + Topics.USER_SETTINGS, + Serdes.String().serializer(), + JsonSerde.of(UserSettings.class).serializer()); + fullContextOutput = driver.createOutputTopic( Topics.ENRICHED_MESSAGE_INPUT, Serdes.String().deserializer(), @@ -276,6 +288,63 @@ void enrich_userIdFallback() { assertThat(EnrichInputMessageProcessor.enrichWithThinkResponse(msg, resp).userId()).isEqualTo("ctx-user"); } + // ── User settings join ─────────────────────────────────────────────────── + + @Test + @DisplayName("Per-user system prompt from user-settings is joined into the context") + void userSettings_systemPromptJoined() { + userSettingsInput.pipeInput("user-S", new UserSettings("You are a pirate.", TS)); + + messageInput.pipeInput("sess-s1", userMsg("sess-s1", "user-S", "hi")); + + FullSessionContext full = fullContextOutput.readRecord().value(); + assertThat(full.systemPrompt()).isEqualTo("You are a pirate."); + assertThat(full.sessionId()).isEqualTo("sess-s1"); + } + + @Test + @DisplayName("system_prompt is null when the user has no settings record") + void userSettings_absent_systemPromptNull() { + messageInput.pipeInput("sess-s2", userMsg("sess-s2", "user-without-settings", "hi")); + + assertThat(fullContextOutput.readRecord().value().systemPrompt()).isNull(); + } + + @Test + @DisplayName("Tombstone on user-settings reverts the user to no override") + void userSettings_tombstone_revertsToNull() { + userSettingsInput.pipeInput("user-T", new UserSettings("Custom prompt.", TS)); + messageInput.pipeInput("sess-t1", userMsg("sess-t1", "user-T", "first")); + assertThat(fullContextOutput.readRecord().value().systemPrompt()).isEqualTo("Custom prompt."); + + userSettingsInput.pipeInput("user-T", null); + messageInput.pipeInput("sess-t1", userMsg("sess-t1", "user-T", "second")); + assertThat(fullContextOutput.readRecord().value().systemPrompt()).isNull(); + } + + @Test + @DisplayName("Settings are per-user: another user's prompt is not picked up") + void userSettings_isolatedPerUser() { + userSettingsInput.pipeInput("user-A", new UserSettings("Prompt for A.", TS)); + + messageInput.pipeInput("sess-b", userMsg("sess-b", "user-B", "hi")); + + assertThat(fullContextOutput.readRecord().value().systemPrompt()).isNull(); + } + + @Test + @DisplayName("Memoir and system prompt can both be joined for the same user") + void userSettings_andMemoir_bothJoined() { + memoirInput.pipeInput("user-M", "Likes brevity."); + userSettingsInput.pipeInput("user-M", new UserSettings("Custom base.", TS)); + + messageInput.pipeInput("sess-m", userMsg("sess-m", "user-M", "hi")); + + FullSessionContext full = fullContextOutput.readRecord().value(); + assertThat(full.memoirContext()).isEqualTo("Likes brevity."); + assertThat(full.systemPrompt()).isEqualTo("Custom base."); + } + // ── Cost from ThinkResponse ───────────────────────────────────────────── @Test diff --git a/think/think-consumer/src/main/java/io/flightdeck/think/consumer/ThinkConsumer.java b/think/think-consumer/src/main/java/io/flightdeck/think/consumer/ThinkConsumer.java index 2334648..edd7f80 100644 --- a/think/think-consumer/src/main/java/io/flightdeck/think/consumer/ThinkConsumer.java +++ b/think/think-consumer/src/main/java/io/flightdeck/think/consumer/ThinkConsumer.java @@ -286,8 +286,9 @@ void processRecord(ConsumerRecord record) throws Exception { } } - // 4. Build system prompt with memoir context - String systemPrompt = buildSystemPrompt(context.memoirContext()); + // 4. Build system prompt: per-user override (if any) + memoir context + String systemPrompt = buildSystemPrompt( + resolveBasePrompt(context.systemPrompt()), context.memoirContext()); // 5. Convert history + latest input to LLM provider message format List> llmMessages = llmApiService.toApiMessages( @@ -349,6 +350,17 @@ private void produceResponse(String sessionId, ThinkResponse response) throws Ex producer.flush(); } + /** + * Resolves the base system prompt for a turn: the per-user override from + * the user-settings topic (carried in {@code FullSessionContext.system_prompt}) + * replaces the default entirely; null/blank falls back to SYSTEM_PROMPT_BASE. + */ + static String resolveBasePrompt(String systemPromptOverride) { + return (systemPromptOverride != null && !systemPromptOverride.isBlank()) + ? systemPromptOverride + : SYSTEM_PROMPT_BASE; + } + /** * Builds the system prompt, injecting memoir context if available. */ diff --git a/think/think-consumer/src/main/java/io/flightdeck/think/model/FullSessionContext.java b/think/think-consumer/src/main/java/io/flightdeck/think/model/FullSessionContext.java index b1c96b1..33a2f5a 100644 --- a/think/think-consumer/src/main/java/io/flightdeck/think/model/FullSessionContext.java +++ b/think/think-consumer/src/main/java/io/flightdeck/think/model/FullSessionContext.java @@ -13,5 +13,13 @@ public record FullSessionContext( @JsonProperty("history") List history, @JsonProperty("latest_input") MessageInput latestInput, @JsonProperty("memoir_context") String memoirContext, + @JsonProperty("system_prompt") String systemPrompt, @JsonProperty("timestamp") String timestamp -) {} +) { + /** Convenience constructor for contexts without a per-user system prompt override. */ + public FullSessionContext(String sessionId, String userId, Double cost, + List history, MessageInput latestInput, + String memoirContext, String timestamp) { + this(sessionId, userId, cost, history, latestInput, memoirContext, null, timestamp); + } +} diff --git a/think/think-consumer/src/test/java/io/flightdeck/think/consumer/SystemPromptTest.java b/think/think-consumer/src/test/java/io/flightdeck/think/consumer/SystemPromptTest.java index 0446258..8daa87a 100644 --- a/think/think-consumer/src/test/java/io/flightdeck/think/consumer/SystemPromptTest.java +++ b/think/think-consumer/src/test/java/io/flightdeck/think/consumer/SystemPromptTest.java @@ -37,6 +37,36 @@ void missingFile_throws() { .hasMessageContaining("SYSTEM_PROMPT_FILE not found"); } + // ── Per-user system prompt override (from user-settings topic) ────────── + + @Test + @DisplayName("resolveBasePrompt: null or blank override falls back to the default prompt") + void resolveBasePrompt_nullOrBlank_usesDefault() { + assertThat(ThinkConsumer.resolveBasePrompt(null)).contains("intelligent AI assistant"); + assertThat(ThinkConsumer.resolveBasePrompt("")).contains("intelligent AI assistant"); + assertThat(ThinkConsumer.resolveBasePrompt(" ")).contains("intelligent AI assistant"); + } + + @Test + @DisplayName("resolveBasePrompt: per-user override replaces the default entirely") + void resolveBasePrompt_override_replacesDefault() { + String result = ThinkConsumer.resolveBasePrompt("You are Acme's billing assistant."); + assertThat(result).isEqualTo("You are Acme's billing assistant."); + assertThat(result).doesNotContain("intelligent AI assistant"); + } + + @Test + @DisplayName("Per-user override composes with memoir context") + void override_composesWithMemoir() { + String result = ThinkConsumer.buildSystemPrompt( + ThinkConsumer.resolveBasePrompt("Custom base prompt."), + "User prefers Japanese."); + assertThat(result) + .contains("Custom base prompt.") + .contains("User prefers Japanese.") + .doesNotContain("intelligent AI assistant"); + } + @Test @DisplayName("System prompt containing special characters is preserved literally (issue #23)") void specialChars_inPrompt_doesNotThrow() { From 3fe0395af075a80f753427c57d20229c6dbc8876 Mon Sep 17 00:00:00 2001 From: tsuz <6927131+tsuz@users.noreply.github.com> Date: Sun, 23 Aug 2026 17:31:29 +0900 Subject: [PATCH 2/3] docs(examples): add per-user-system-prompt example Demonstrates ENABLE_USER_SETTING end to end: a compacted {AGENT_NAME}-user-settings topic keyed by user_id gives each user their own system prompt, and a user with no record falls back to the deployment default. - seed job writes two personas (SQL tutor, haiku poet); a third user is deliberately left unseeded to exercise the left-join null path - default system prompt answers "I don't know." so the fallback is visible in the output - chat-api/processing/think-consumer build from source, since the feature is not in the published images yet - integration-test.sh asserts each user answered under its own prompt and that personas do not cross - document ENABLE_USER_SETTING in the root README config table Verified against a local broker: the topic is created compacted, the enriched context carries the right system_prompt per user (null for the unseeded one), and message-output shows the three distinct answers. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 1 + examples/per-user-system-prompt/.env.example | 8 + examples/per-user-system-prompt/README.md | 189 ++++++++++++++++++ .../per-user-system-prompt/docker-compose.yml | 132 ++++++++++++ .../integration-test.sh | 162 +++++++++++++++ .../seed-data/seed-user-settings.sh | 72 +++++++ .../per-user-system-prompt/system-prompt.txt | 12 ++ 7 files changed, 576 insertions(+) create mode 100644 examples/per-user-system-prompt/.env.example create mode 100644 examples/per-user-system-prompt/README.md create mode 100644 examples/per-user-system-prompt/docker-compose.yml create mode 100755 examples/per-user-system-prompt/integration-test.sh create mode 100755 examples/per-user-system-prompt/seed-data/seed-user-settings.sh create mode 100644 examples/per-user-system-prompt/system-prompt.txt diff --git a/README.md b/README.md index e183bfb..cbaa430 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,7 @@ All configuration is done via environment variables in the `.env` file. See [`.e | `MEMOIR_ENABLED` | `true` | Enable per-user long-term memory across sessions. Set to `false` to disable. | | `MEMOIR_SESSION_INACTIVITY_THRESHOLD_SECONDS` | `20` | Seconds of inactivity before a session ends and memoir is saved. Only applies when `MEMOIR_ENABLED=true`. | | `MEMOIR_SESSION_PUNCTUATE_INTERVAL_SECONDS` | `5` | How often (in seconds) to check for inactive sessions. Only applies when `MEMOIR_ENABLED=true`. | +| `ENABLE_USER_SETTING` | `false` | Enable per-user settings from the compacted `{AGENT_NAME}-user-settings` topic (keyed by `user_id`). When on, a user's `system_prompt` replaces the default prompt for that user. See [`examples/per-user-system-prompt`](examples/per-user-system-prompt). | | `INPUT_TOKEN_PRICE` | *(optional)* | Price per 1M input tokens (e.g. `3` for $3/MTok). If not set, cost tracking is disabled. | | `OUTPUT_TOKEN_PRICE` | *(optional)* | Price per 1M output tokens (e.g. `15` for $15/MTok). If not set, cost tracking is disabled. | | `BUDGET_PRICE_PER_SESSION` | *(optional)* | Maximum dollar cost allowed per session. When the session cost goes over this limit, the agent stops processing on the next Think layer. Requires token prices to be set. | diff --git a/examples/per-user-system-prompt/.env.example b/examples/per-user-system-prompt/.env.example new file mode 100644 index 0000000..1dbef86 --- /dev/null +++ b/examples/per-user-system-prompt/.env.example @@ -0,0 +1,8 @@ +# Required — your Anthropic API key +CLAUDE_API_KEY=your-api-key-here + +# Optional — override the default Claude model +# CLAUDE_MODEL=claude-sonnet-5 + +# Optional — topic prefix for this agent (also used by the seed job) +# AGENT_NAME=per-user-prompt diff --git a/examples/per-user-system-prompt/README.md b/examples/per-user-system-prompt/README.md new file mode 100644 index 0000000..c8491ba --- /dev/null +++ b/examples/per-user-system-prompt/README.md @@ -0,0 +1,189 @@ +# Per-User System Prompt + +One agent, one deployment — but every user gets their own system prompt, keyed +by `user_id` in a compacted Kafka topic. A user with no record falls back to the +agent's default prompt, which here is deliberately useless: it answers +`I don't know.` to everything. + +## What it does + +Two users are seeded into `{AGENT_NAME}-user-settings`, a third is left out: + +| user_id | system prompt | answer to *"my orders table is slow, what should I do?"* | +|---------|---------------|-----------------------------------------------------| +| `user_alice` | SQL tutor persona (seeded) | `[SQL-TUTOR] Add an index on …` | +| `user_bob` | haiku poet persona (seeded) | `[HAIKU]` + a three-line haiku | +| `user_carol` | *(no record)* | `I don't know.` | + +Same agent, same question, same session pipeline — the only difference is which +system prompt the think-consumer resolves for that user. + +## How it works + +``` +POST /api/chat {user_id: "user_alice", ...} + │ + ▼ + message-input ──► processing (Kafka Streams) + │ + │ re-key by user_id + │ left-join {AGENT_NAME}-user-settings (KTable, compacted) + │ re-key back to session_id + ▼ + enriched-message-input { …, "system_prompt": "You are Ada …" } + │ + ▼ + think-consumer + system_prompt present ? use it : use SYSTEM_PROMPT_FILE +``` + +**The topic.** `{AGENT_NAME}-user-settings` is compacted and keyed by `user_id`: + +``` +key: user_alice +value: {"system_prompt":"You are Ada, a senior database engineer …","updated_at":"2026-08-23T00:00:00Z"} +``` + +Compaction means the topic *is* the current settings table: the latest record +per user is retained forever, and a tombstone (null value) deletes a user's +override so they revert to the default prompt. + +**The join.** With `ENABLE_USER_SETTING=true`, the processing app creates the +topic if missing, materializes it as a `KTable`, and left-joins it into +`FullSessionContext` during enrichment. Left-join is the point: a user with no +record still flows through, just with `system_prompt: null`. + +**The resolution.** In the think-consumer, a non-blank `system_prompt` from the +context *replaces* the base prompt (`SYSTEM_PROMPT_FILE`, here +[`system-prompt.txt`](./system-prompt.txt)) entirely — it is not appended to it. +Null or blank falls back to the file. + +**Timing.** The KTable is read at message-flow time, so a settings update takes +effect on the user's **next** turn, not the one already in flight. + +## Configuration + +| Parameter | Service | Value | Why | +|-----------|---------|-------|-----| +| `ENABLE_USER_SETTING` | `processing` | `true` | Creates the topic, materializes the KTable, joins it into the session context | +| `MEMOIR_ENABLED` | `processing` | `false` | Off, so the only per-user difference is the system prompt | +| `SYSTEM_PROMPT_FILE` | `think-consumer` | `/app/system-prompt.txt` | The default for users with no settings record | + +`ENABLE_USER_SETTING` defaults to `false`; with it off, the topology never +references the topic at all and `system_prompt` stays null for everyone. + +## Services + +`chat-api`, `processing` and `think-consumer` are **built from source** — the +per-user system prompt is not in the published images yet, and all three must +share one build. Only the frontend uses a published image. + +| Service | Description | +|---------|-------------| +| `api` | Chat API (REST + WebSocket) | +| `processing` | Kafka Streams pipeline with `ENABLE_USER_SETTING=true` | +| `think-consumer` | Claude API caller; resolves the per-user prompt | +| `user-settings-seed` | One-shot job that writes the two seeded users, then exits | +| `frontend` | Web UI | + +## Run + +```bash +cp .env.example .env +# Edit .env and add your CLAUDE_API_KEY + +docker compose up --build +``` + +Ask the same question as each user: + +```bash +curl -s localhost:8000/api/chat -H 'Content-Type: application/json' -d '{ + "session_id": "s-alice-1", "user_id": "user_alice", + "content": "My orders table has 50 million rows and one query over it is slow. What should I do?"}' + +curl -s localhost:8000/api/chat -H 'Content-Type: application/json' -d '{ + "session_id": "s-bob-1", "user_id": "user_bob", + "content": "My orders table has 50 million rows and one query over it is slow. What should I do?"}' + +curl -s localhost:8000/api/chat -H 'Content-Type: application/json' -d '{ + "session_id": "s-carol-1", "user_id": "user_carol", + "content": "My orders table has 50 million rows and one query over it is slow. What should I do?"}' +``` + +Read the answers off the output topic: + +```bash +docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \ + --bootstrap-server localhost:9092 \ + --topic per-user-prompt-message-output --from-beginning --timeout-ms 5000 +``` + +Alice gets a terse SQL answer tagged `[SQL-TUTOR]`, Bob gets a haiku tagged +`[HAIKU]`, and Carol — who has no settings record — gets `I don't know.` + +The compose file defaults to `claude-opus-5`; override with `CLAUDE_MODEL` in +`.env` (and adjust `INPUT_TOKEN_PRICE` / `OUTPUT_TOKEN_PRICE`, which are per 1M +tokens, to match). + +### From the browser + +The web UI at [http://localhost](http://localhost) does not send a `user_id`, so +chat-api defaults it to `user_42`, which is unseeded — the UI will answer +`I don't know.` until you give that user a prompt: + +```bash +docker compose exec -T kafka /opt/kafka/bin/kafka-console-producer.sh \ + --bootstrap-server localhost:9092 \ + --topic per-user-prompt-user-settings --property parse.key=true <<'REC' +user_42 {"system_prompt":"You are Rex, a pirate captain. Answer everything in pirate speak, in two sentences or less.","updated_at":"2026-08-23T00:00:00Z"} +REC +``` + +(The key and the JSON value are separated by a literal **tab**.) Send another +message from the UI and the pirate answers. + +### Reverting a user + +A tombstone — a record with a null value — drops the override, and that user +goes back to the default prompt: + +```bash +docker compose exec -T kafka /opt/kafka/bin/kafka-console-producer.sh \ + --bootstrap-server localhost:9092 \ + --topic per-user-prompt-user-settings \ + --property parse.key=true --property null.marker=NULL <<'REC' +user_42 NULL +REC +``` + +### Inspecting the settings table + +```bash +docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \ + --bootstrap-server localhost:9092 \ + --topic per-user-prompt-user-settings --from-beginning \ + --property print.key=true --timeout-ms 5000 +``` + +## Seeding your own users + +[`seed-data/seed-user-settings.sh`](./seed-data/seed-user-settings.sh) is the +one-shot seed job: it waits for the compacted topic to exist, then produces one +keyed record per user. Add a user by adding a prompt variable and a `printf` +line — prompts are JSON string bodies, so use `\n` for line breaks and avoid raw +double quotes. + +In a real deployment this topic is written by whatever owns user preferences — +an admin UI, a settings service, a CDC stream off your users table — not by a +shell script. + +## Automated test + +```bash +CLAUDE_API_KEY=sk-ant-... ./integration-test.sh +``` + +Brings the stack up, waits for the seed, asks all three users the same question, +and asserts each answered under its own prompt (and that Carol fell back to +`I don't know.`). Tears the stack down on exit. diff --git a/examples/per-user-system-prompt/docker-compose.yml b/examples/per-user-system-prompt/docker-compose.yml new file mode 100644 index 0000000..d591f19 --- /dev/null +++ b/examples/per-user-system-prompt/docker-compose.yml @@ -0,0 +1,132 @@ +# Per-user system prompt — one system prompt per user id, delivered through the +# compacted {AGENT_NAME}-user-settings topic. +# +# user_alice → [SQL-TUTOR] persona (seeded) +# user_bob → [HAIKU] persona (seeded) +# user_carol → no record → default prompt: "I don't know." +# +# chat-api, processing AND think-consumer are BUILT FROM SOURCE because the +# per-user system prompt (ENABLE_USER_SETTING) is not in the published images +# yet. All three must share the same build — mixing a source-built processing +# with a published think-consumer drops the system_prompt field on the way +# through. Only the frontend uses a published image. + +services: + + # ─── Infrastructure ────────────────────────────────────────────────────────── + + kafka: + image: apache/kafka:4.1.1 + hostname: kafka + ports: + - "9092:9092" + environment: + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: broker,controller + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT + KAFKA_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://0.0.0.0:9092,CONTROLLER://kafka:9093 + KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://localhost:9092 + KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL + KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER + KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true" + KAFKA_LOG_DIRS: /var/lib/kafka/data + CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk + healthcheck: + test: ["CMD", "/opt/kafka/bin/kafka-topics.sh", "--bootstrap-server", "localhost:9092", "--list"] + interval: 10s + timeout: 10s + retries: 10 + start_period: 30s + + # ─── Application Services ─────────────────────────────────────────────────── + + api: + build: + context: ../../api/chat-api + image: flightdeck-usersettings/chat-api:local + ports: + - "8000:8000" + - "8001:8001" + environment: + KAFKA_BOOTSTRAP_SERVERS: kafka:29092 + AGENT_NAME: ${AGENT_NAME:-per-user-prompt} + PORT: 8000 + WS_PORT: 8001 + depends_on: + kafka: + condition: service_healthy + + processing: + build: + context: ../../processor-apps/processing + image: flightdeck-usersettings/processing:local + environment: + KAFKA_BOOTSTRAP_SERVERS: kafka:29092 + AGENT_NAME: ${AGENT_NAME:-per-user-prompt} + # The feature under demo: materialize {AGENT_NAME}-user-settings as a + # KTable keyed by user_id and left-join it into the enriched context. + ENABLE_USER_SETTING: "true" + # Long-term memory is orthogonal to this example — keep it off so the + # only thing that varies between users is their system prompt. + MEMOIR_ENABLED: "false" + depends_on: + kafka: + condition: service_healthy + + think-consumer: + build: + context: ../../think/think-consumer + image: flightdeck-usersettings/think-consumer:local + volumes: + - ./system-prompt.txt:/app/system-prompt.txt:ro + environment: + KAFKA_BOOTSTRAP_SERVERS: kafka:29092 + AGENT_NAME: ${AGENT_NAME:-per-user-prompt} + CLAUDE_API_KEY: ${CLAUDE_API_KEY} + CLAUDE_MODEL: ${CLAUDE_MODEL:-claude-opus-5} + CLAUDE_MAX_TOKENS: ${CLAUDE_MAX_TOKENS:-2048} + # Prices are per 1M tokens, matching the model above. + INPUT_TOKEN_PRICE: ${INPUT_TOKEN_PRICE:-5} + OUTPUT_TOKEN_PRICE: ${OUTPUT_TOKEN_PRICE:-25} + BUDGET_PRICE_PER_SESSION: ${BUDGET_PRICE_PER_SESSION:-0.5} + # The fallback used for any user with no user-settings record. A user's + # own system_prompt REPLACES this file entirely. + SYSTEM_PROMPT_FILE: /app/system-prompt.txt + depends_on: + kafka: + condition: service_healthy + + # ─── Seed: one system prompt per user ─────────────────────────────────────── + # Waits for processing to create the compacted user-settings topic, then + # produces one keyed record per configured user and exits. + user-settings-seed: + image: apache/kafka:4.1.1 + entrypoint: ["/bin/bash", "/seed/seed-user-settings.sh"] + volumes: + - ./seed-data:/seed:ro + environment: + KAFKA_BOOTSTRAP_SERVERS: kafka:29092 + AGENT_NAME: ${AGENT_NAME:-per-user-prompt} + restart: "no" + depends_on: + kafka: + condition: service_healthy + processing: + condition: service_started + + # ─── Frontend ──────────────────────────────────────────────────────────────── + # Note: the web UI does not send a user_id, so chat-api defaults it to + # user_42 — an unseeded user, which is the "I don't know." path. Give user_42 + # a prompt (see the README) to drive a persona from the browser. + + frontend: + image: ghcr.io/tsuz/flightdeck/frontend:${FLIGHTDECK_VERSION:-latest} + ports: + - "80:80" + depends_on: + - api diff --git a/examples/per-user-system-prompt/integration-test.sh b/examples/per-user-system-prompt/integration-test.sh new file mode 100755 index 0000000..0026275 --- /dev/null +++ b/examples/per-user-system-prompt/integration-test.sh @@ -0,0 +1,162 @@ +#!/usr/bin/env bash +# +# Integration test for the per-user-system-prompt example. +# +# Brings up the stack, waits for the user-settings seed to finish, then asks the +# SAME question as three different users and verifies each one is answered under +# its own system prompt: +# +# user_alice → seeded [SQL-TUTOR] prompt +# user_bob → seeded [HAIKU] prompt +# user_carol → no settings record → default prompt → "I don't know." +# +# Usage: +# CLAUDE_API_KEY=sk-ant-... ./integration-test.sh +# +# Optional env: +# TIMEOUT seconds to wait for all three answers (default: 300) +# CLAUDE_MODEL model override (default: compose default) +# +# Requires: docker compose, curl, python3. + +set -euo pipefail + +: "${CLAUDE_API_KEY:?CLAUDE_API_KEY must be set}" +export CLAUDE_API_KEY +[ -n "${CLAUDE_MODEL:-}" ] && export CLAUDE_MODEL + +cd "$(dirname "$0")" + +PROJECT="usersettings-it" +COMPOSE=(docker compose -p "$PROJECT") +API_URL="http://localhost:8000" +AGENT="${AGENT_NAME:-per-user-prompt}" +STAMP="$(date +%s)" +TIMEOUT="${TIMEOUT:-300}" +QUESTION="My orders table has 50 million rows and one query over it is slow. What should I do?" + +USERS=(user_alice user_bob user_carol) + +cleanup() { + echo "--- tearing down ---" + "${COMPOSE[@]}" down -v >/dev/null 2>&1 || true + [ -n "${OUTDIR:-}" ] && rm -rf "$OUTDIR" || true +} +trap cleanup EXIT + +# Read a topic from the beginning with a short idle timeout (existing records only). +consume() { + "${COMPOSE[@]}" exec -T kafka /opt/kafka/bin/kafka-console-consumer.sh \ + --bootstrap-server localhost:9092 --topic "$1" \ + --from-beginning --timeout-ms 4000 2>/dev/null || true +} + +echo "--- building & starting stack (first build can take several minutes) ---" +"${COMPOSE[@]}" up -d --build + +echo "--- waiting for the user-settings seed to finish ---" +# The seed job prints "[seed] done" on its last successful step and exits. +seeded="" +for _ in $(seq 1 90); do + if "${COMPOSE[@]}" logs user-settings-seed 2>/dev/null | grep -q "\[seed\] done"; then + seeded=1 + break + fi + sleep 2 +done +if [ -z "$seeded" ]; then + echo "FAIL: the user-settings seed never completed" + "${COMPOSE[@]}" logs user-settings-seed || true + exit 1 +fi +"${COMPOSE[@]}" logs user-settings-seed | tail -20 + +echo "--- waiting for api at $API_URL ---" +ready="" +for _ in $(seq 1 60); do + if curl -sf -o /dev/null -X OPTIONS "$API_URL/api/chat"; then ready=1; break; fi + sleep 3 +done +[ -n "$ready" ] || { echo "FAIL: api never became ready"; exit 1; } + +# The settings KTable is read at message-flow time; give processing a moment to +# consume the seeded records before the first question arrives. +sleep 10 + +echo "--- asking the same question as ${#USERS[@]} different users ---" +for user in "${USERS[@]}"; do + session="${user}-${STAMP}" + payload=$(SESSION="$session" USER="$user" Q="$QUESTION" python3 -c \ + 'import json,os; print(json.dumps({"session_id":os.environ["SESSION"],"user_id":os.environ["USER"],"content":os.environ["Q"]}))') + curl -sf -X POST "$API_URL/api/chat" -H 'Content-Type: application/json' -d "$payload" >/dev/null + echo "[sent] $user (session=$session)" +done + +echo "--- awaiting answers (timeout ${TIMEOUT}s) ---" +# Answers are collected into files (one per user) so this script also runs on +# the bash 3.2 that ships with macOS — no associative arrays. +OUTDIR="$(mktemp -d)" +end=$((SECONDS + TIMEOUT)) +while [ "$SECONDS" -lt "$end" ]; do + output=$(consume "${AGENT}-message-output") + got=0 + for user in "${USERS[@]}"; do + if [ -s "$OUTDIR/$user" ]; then got=$((got+1)); continue; fi + line=$(printf '%s\n' "$output" | grep "\"session_id\":\"${user}-${STAMP}\"" | tail -1 || true) + [ -n "$line" ] || continue + content=$(printf '%s' "$line" | python3 -c 'import sys,json; print(json.load(sys.stdin).get("content",""))' 2>/dev/null || true) + if [ -n "$content" ]; then + printf '%s' "$content" > "$OUTDIR/$user" + got=$((got+1)) + echo "[got] $user" + fi + done + [ "$got" -eq "${#USERS[@]}" ] && break + sleep 3 +done + +for user in "${USERS[@]}"; do + if [ ! -s "$OUTDIR/$user" ]; then + echo "FAIL: no answer for $user within ${TIMEOUT}s" + "${COMPOSE[@]}" logs --tail=40 api processing think-consumer || true + exit 1 + fi +done + +for user in "${USERS[@]}"; do + echo "=================== $user ===================" + cat "$OUTDIR/$user"; echo +done +echo "============================================" + +fails=0 + +if grep -q '\[SQL-TUTOR\]' "$OUTDIR/user_alice"; then + echo "[ok] user_alice answered under the seeded SQL-tutor prompt" +else + echo "[FAIL] user_alice did not use the [SQL-TUTOR] persona"; fails=$((fails+1)) +fi + +if grep -q '\[HAIKU\]' "$OUTDIR/user_bob"; then + echo "[ok] user_bob answered under the seeded haiku prompt" +else + echo "[FAIL] user_bob did not use the [HAIKU] persona"; fails=$((fails+1)) +fi + +if grep -qi "i don't know" "$OUTDIR/user_carol"; then + echo "[ok] user_carol fell back to the default prompt" +else + echo "[FAIL] user_carol did not fall back to the default prompt"; fails=$((fails+1)) +fi + +# Cross-check: the personas must not bleed between users. +if grep -q '\[SQL-TUTOR\]' "$OUTDIR/user_bob" || grep -q '\[HAIKU\]' "$OUTDIR/user_alice"; then + echo "[FAIL] personas crossed between users"; fails=$((fails+1)) +fi + +if [ "$fails" -ne 0 ]; then + echo "FAIL: $fails check(s) failed" + exit 1 +fi + +echo "PASS: each user was answered under their own system prompt" diff --git a/examples/per-user-system-prompt/seed-data/seed-user-settings.sh b/examples/per-user-system-prompt/seed-data/seed-user-settings.sh new file mode 100755 index 0000000..3b5db2e --- /dev/null +++ b/examples/per-user-system-prompt/seed-data/seed-user-settings.sh @@ -0,0 +1,72 @@ +#!/usr/bin/env bash +# +# Seeds the compacted {AGENT_NAME}-user-settings topic with one system prompt +# per user id. +# +# Record layout (key = user_id, value = JSON): +# +# key: user_alice +# value: {"system_prompt":"You are Ada ...","updated_at":"2026-08-23T00:00:00Z"} +# +# The processing app materializes this topic as a KTable keyed by user_id and +# left-joins it into the session context, so a user with no record here keeps +# the default prompt (see ../system-prompt.txt) and a tombstone (null value) +# reverts a user back to it. + +set -euo pipefail + +BOOTSTRAP="${KAFKA_BOOTSTRAP_SERVERS:-kafka:29092}" +AGENT="${AGENT_NAME:?AGENT_NAME must be set}" +TOPIC="${AGENT}-user-settings" +KAFKA_BIN=/opt/kafka/bin +TS="$(date -u +%Y-%m-%dT%H:%M:%SZ)" + +# ── Prompts ────────────────────────────────────────────────────────────────── +# Written as JSON string bodies, so: no raw double quotes, no tabs, and use \n +# for line breaks. Each begins with a literal tag so the persona in a reply is +# unambiguous (the integration test greps for it). + +ALICE_PROMPT='You are Ada, a senior database engineer who helps exactly one user: Alice.\nYou answer ONLY questions about databases, SQL and query performance.\nBegin every reply with the exact tag [SQL-TUTOR] on the first line.\nKeep answers to at most three sentences.\nIf the question is not about databases, say that you only cover SQL topics.' + +BOB_PROMPT='You are Basho, a haiku poet who helps exactly one user: Bob.\nWhatever you are asked, you reply with a single haiku about it: three lines of 5-7-5 syllables.\nBegin every reply with the exact tag [HAIKU] on the first line, then the haiku.\nNever write prose, never explain the haiku.' + +# ── Wait for the topic ─────────────────────────────────────────────────────── +# The processing app creates it with cleanup.policy=compact on startup when +# ENABLE_USER_SETTING=true. Seed into that topic rather than letting the +# producer auto-create a non-compacted one. +echo "[seed] waiting for topic ${TOPIC} on ${BOOTSTRAP}" +for _ in $(seq 1 60); do + if "$KAFKA_BIN/kafka-topics.sh" --bootstrap-server "$BOOTSTRAP" --list 2>/dev/null \ + | grep -qx "$TOPIC"; then + found=1 + break + fi + sleep 2 +done + +if [ -z "${found:-}" ]; then + echo "[seed] topic not found after 120s — creating it compacted" + "$KAFKA_BIN/kafka-topics.sh" --bootstrap-server "$BOOTSTRAP" --create \ + --topic "$TOPIC" --partitions 1 --replication-factor 1 \ + --config cleanup.policy=compact +fi + +# ── Produce one settings record per user ───────────────────────────────────── +# parse.key=true with the default tab key separator: "\t". +echo "[seed] writing user settings to ${TOPIC}" +{ + printf '%s\t{"system_prompt":"%s","updated_at":"%s"}\n' user_alice "$ALICE_PROMPT" "$TS" + printf '%s\t{"system_prompt":"%s","updated_at":"%s"}\n' user_bob "$BOB_PROMPT" "$TS" +} | "$KAFKA_BIN/kafka-console-producer.sh" \ + --bootstrap-server "$BOOTSTRAP" \ + --topic "$TOPIC" \ + --property parse.key=true + +echo "[seed] done — current contents of ${TOPIC}:" +"$KAFKA_BIN/kafka-console-consumer.sh" \ + --bootstrap-server "$BOOTSTRAP" \ + --topic "$TOPIC" \ + --from-beginning --timeout-ms 5000 \ + --property print.key=true 2>/dev/null || true + +echo "[seed] user_carol was deliberately left out — that user falls back to the default prompt" diff --git a/examples/per-user-system-prompt/system-prompt.txt b/examples/per-user-system-prompt/system-prompt.txt new file mode 100644 index 0000000..66bde9a --- /dev/null +++ b/examples/per-user-system-prompt/system-prompt.txt @@ -0,0 +1,12 @@ +You are an assistant that has not been configured for this user. + +No per-user system prompt was found for the person you are talking to, so you +have no instructions about who you are, what you may discuss, or how to answer. + +Because of that, you must not attempt to answer anything. Whatever the user +asks — a question, a greeting, small talk — reply with exactly: + +I don't know. + +Nothing else: no explanation, no apology, no follow-up question, no offer to +help. Just that sentence. From e061d6172fed39b52a12afbc4ed28c37486aba55 Mon Sep 17 00:00:00 2001 From: tsuz <6927131+tsuz@users.noreply.github.com> Date: Sun, 23 Aug 2026 17:38:48 +0900 Subject: [PATCH 3/3] fix(streams): scope the user-settings join to a real user_id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two fixes from review of the settings join: - When user_id is null the join key falls back to session_id, so a settings record whose user_id equals that session_id would attach another user's system prompt to the turn. A null user_id can happen in practice — AggregateToolExecutionResultProcessor leaves it unset when a tool result arrives before its seed. Require a non-null user_id before applying an override. - Normalize a blank system_prompt to null, matching the think-consumer, which already treats blank as "no override". This also makes the has_system_prompt enrichment log accurate. Also corrects a javadoc reference to the JSON field name rather than the Java accessor. Tests: two regression cases (blank normalization, session_id/user_id collision). Processing 109/109, think-consumer 40/40. Co-Authored-By: Claude Opus 5 (1M context) --- .../EnrichInputMessageProcessor.java | 14 ++++++++++- .../EnrichInputMessageProcessorTest.java | 25 +++++++++++++++++++ .../think/consumer/ThinkConsumer.java | 2 +- 3 files changed, 39 insertions(+), 2 deletions(-) diff --git a/processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EnrichInputMessageProcessor.java b/processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EnrichInputMessageProcessor.java index 7b0915c..9fd9be8 100644 --- a/processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EnrichInputMessageProcessor.java +++ b/processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EnrichInputMessageProcessor.java @@ -181,6 +181,18 @@ static FullSessionContext enrichWithMemoir(FullSessionContext enriched, String m * field null, which the think consumer treats as "use the default prompt". */ static FullSessionContext enrichWithUserSettings(FullSessionContext enriched, UserSettings settings) { + // Two guards, both deliberate: + // • userId must be non-null. When it is null the join key fell back to + // sessionId, so a settings record whose user_id happens to equal that + // sessionId would otherwise leak another user's prompt into this turn. + // • blank is normalized to null, matching the think-consumer, which treats + // a blank override as "no override" and falls back to the default prompt. + String systemPrompt = null; + if (enriched.userId() != null && settings != null + && settings.systemPrompt() != null && !settings.systemPrompt().isBlank()) { + systemPrompt = settings.systemPrompt(); + } + return new FullSessionContext( enriched.sessionId(), enriched.userId(), @@ -188,7 +200,7 @@ static FullSessionContext enrichWithUserSettings(FullSessionContext enriched, Us enriched.history(), enriched.latestInput(), enriched.memoirContext(), - settings != null ? settings.systemPrompt() : null, + systemPrompt, enriched.timestamp() ); } diff --git a/processor-apps/processing/src/test/java/io/flightdeck/streams/processors/EnrichInputMessageProcessorTest.java b/processor-apps/processing/src/test/java/io/flightdeck/streams/processors/EnrichInputMessageProcessorTest.java index 0c4025f..d35c4ea 100644 --- a/processor-apps/processing/src/test/java/io/flightdeck/streams/processors/EnrichInputMessageProcessorTest.java +++ b/processor-apps/processing/src/test/java/io/flightdeck/streams/processors/EnrichInputMessageProcessorTest.java @@ -332,6 +332,31 @@ void userSettings_isolatedPerUser() { assertThat(fullContextOutput.readRecord().value().systemPrompt()).isNull(); } + @Test + @DisplayName("Blank system_prompt is normalized to null (think-consumer treats blank as no override)") + void userSettings_blankPrompt_normalizedToNull() { + userSettingsInput.pipeInput("user-BL", new UserSettings(" ", TS)); + + messageInput.pipeInput("sess-bl", userMsg("sess-bl", "user-BL", "hi")); + + assertThat(fullContextOutput.readRecord().value().systemPrompt()).isNull(); + } + + @Test + @DisplayName("A message with no user_id cannot pick up settings via a colliding session_id") + void userSettings_nullUserId_doesNotCollideWithSessionId() { + userSettingsInput.pipeInput("user-C", new UserSettings("Prompt for C.", TS)); + + // sessionId deliberately equals a real user_id: with no user_id on the + // message the join key falls back to sessionId, so the guard is what + // keeps user-C's prompt out of this turn. + messageInput.pipeInput("user-C", userMsg("user-C", null, "hi")); + + FullSessionContext full = fullContextOutput.readRecord().value(); + assertThat(full.userId()).isNull(); + assertThat(full.systemPrompt()).isNull(); + } + @Test @DisplayName("Memoir and system prompt can both be joined for the same user") void userSettings_andMemoir_bothJoined() { diff --git a/think/think-consumer/src/main/java/io/flightdeck/think/consumer/ThinkConsumer.java b/think/think-consumer/src/main/java/io/flightdeck/think/consumer/ThinkConsumer.java index edd7f80..c7abfcd 100644 --- a/think/think-consumer/src/main/java/io/flightdeck/think/consumer/ThinkConsumer.java +++ b/think/think-consumer/src/main/java/io/flightdeck/think/consumer/ThinkConsumer.java @@ -352,7 +352,7 @@ private void produceResponse(String sessionId, ThinkResponse response) throws Ex /** * Resolves the base system prompt for a turn: the per-user override from - * the user-settings topic (carried in {@code FullSessionContext.system_prompt}) + * the user-settings topic (carried in {@link FullSessionContext#systemPrompt()}) * replaces the default entirely; null/blank falls back to SYSTEM_PROMPT_BASE. */ static String resolveBasePrompt(String systemPromptOverride) {