bobbai00 commented on code in PR #8753:
URL: https://github.com/apache/texera/pull/8753#discussion_r4319074026
##########
agent-service/src/server.ts:
##########
@@ -493,6 +493,15 @@ export function buildApp() {
wsLog.info({ agentId, preview: msg.content.substring(0, 50) },
"received command");
+ // Reject a prompt that arrives while a run is already in flight
+ // instead of re-pointing that run's warehouse mid-execution
(#8711).
+ // Checked before setDelegateWarehouse so a busy agent's delegate
+ // config is left untouched.
+ if (agent.getState() !== AgentState.AVAILABLE) {
+ sendEventToClient(ws, new WsServerErrorEvent("Agent is busy
processing another message"));
Review Comment:
**Must fix:**
The operator-feedback panel, which #8711 names as the reachable mid-run
sender, still sends without checking agent state
(`agent-interaction.component.ts:175-177`). It then shows a success toast and
clears the textarea (`agent-interaction.component.ts:161-162`). So a rejected
prompt leaves a green toast, a red busy toast, and lost input. I'd gate its
`canSend()` on `AVAILABLE` in this PR, as the chat panel does
(`agent-chat.component.ts:382-384`).
##########
agent-service/src/agent/texera-agent.ts:
##########
@@ -498,17 +498,23 @@ export class TexeraAgent {
}
async sendMessage(userMessage: string, messageSource?: "chat" | "feedback"):
Promise<AgentMessageResult> {
- const messageId =
`msg-${this.agentId}-${++this.messageCounter}-${Date.now()}`;
- let stepIndex = 0;
-
- await this.refreshWorkflowFromBackend();
-
+ // State flips synchronously, before the first await below, so a second
+ // prompt arriving while this one is mid-flight (e.g. during
+ // refreshWorkflowFromBackend) sees GENERATING and is rejected instead of
+ // interleaving with this run's head/abortController/currentMessageId
(#8711).
+ if (this.state !== AgentStateEnum.AVAILABLE) {
+ throw new Error("Agent is busy processing another message");
+ }
+ this.state = AgentStateEnum.GENERATING;
this.abortController = new AbortController();
- this.state = AgentStateEnum.GENERATING;
+ const messageId =
`msg-${this.agentId}-${++this.messageCounter}-${Date.now()}`;
+ let stepIndex = 0;
this.currentMessageId = messageId;
+ await this.refreshWorkflowFromBackend();
Review Comment:
**Advisory:**
With state flipped above, this await is outside the `try` whose `finally`
(line 739) resets it, so a throw here would wedge the agent.
`refreshWorkflowFromBackend` catches everything today (lines 414-421), so it is
not reachable yet. Moving the await inside the `try` makes the reset structural.
##########
agent-service/src/agent/texera-agent.ts:
##########
@@ -498,17 +498,23 @@ export class TexeraAgent {
}
async sendMessage(userMessage: string, messageSource?: "chat" | "feedback"):
Promise<AgentMessageResult> {
- const messageId =
`msg-${this.agentId}-${++this.messageCounter}-${Date.now()}`;
- let stepIndex = 0;
-
- await this.refreshWorkflowFromBackend();
-
+ // State flips synchronously, before the first await below, so a second
+ // prompt arriving while this one is mid-flight (e.g. during
+ // refreshWorkflowFromBackend) sees GENERATING and is rejected instead of
+ // interleaving with this run's head/abortController/currentMessageId
(#8711).
+ if (this.state !== AgentStateEnum.AVAILABLE) {
Review Comment:
**Must fix:**
This guard makes an idle stop permanent. `stop()` sets `STOPPING`
unconditionally (line 756), and only this method's `finally` (line 742) resets
it. The guard never reaches that `finally`, so every later prompt is rejected
until the agent is deleted (reproduced with a scratch test). `POST
/agents/:id/stop` reaches this (`server.ts:303-307`), as does a Stop click
landing just after a run ends. I'd return early from `stop()` unless state is
`GENERATING`, with a test.
##########
agent-service/src/agent/texera-agent.spec.ts:
##########
@@ -817,6 +817,32 @@ describe("sendMessage", () => {
expect(observed).toEqual([AgentState.GENERATING]);
expect(agent.getState()).toBe(AgentState.AVAILABLE);
});
+
+ test("a second sendMessage while one is in flight is rejected, not
interleaved (#8711)", async () => {
+ let release!: () => void;
+ const gate = new Promise<void>(r => (release = r));
+ const model = new MockLanguageModelV4({
+ doGenerate: async () => {
+ await gate;
+ return {
+ content: [{ type: "text", text: "x" }],
+ finishReason: finish("stop"),
+ usage: usage(1, 1),
+ warnings: [],
+ } as any;
+ },
+ });
+ const agent = makeAgentWith(model);
+ const first = agent.sendMessage("first");
+ await new Promise(r => setTimeout(r, 5));
Review Comment:
**Advisory:**
With this sleep, the test still passes when the refresh await moves back
above the state flip. So it pins the guard but not the ordering fix. Without
the sleep it passes here, and I checked that it fails with the old ordering.
```suggestion
```
##########
agent-service/src/server.ws.spec.ts:
##########
@@ -299,6 +299,48 @@ describe(`WS ${API}/agents/:id/react`, () => {
expect(resting.state).toBe("AVAILABLE");
});
+ test("a prompt sent while a run is in flight is rejected, not interleaved
(#8711)", async () => {
+ const id = await createAgent();
+ const agent = _getAgentForTests(id)! as any;
+
+ // Pin an initial warehouse so we can assert the busy-rejected prompt below
+ // never reaches setDelegateWarehouse and re-points it mid-run (#8711).
+ agent.setDelegateConfig({ userToken: "t", workflowId: 1, warehouseId: 7 });
+
+ let release!: () => void;
+ const gate = new Promise<void>(r => (release = r));
+ let callCount = 0;
+ agent.sendMessage = async function (this: any) {
+ callCount++;
+ this.state = "GENERATING";
+ await gate;
+ this.state = "AVAILABLE";
+ return { response: "done", messages: [], usage: { inputTokens: 0,
outputTokens: 0, totalTokens: 0 }, stopped: false };
Review Comment:
**Must fix:**
This line is over Prettier's `printWidth` of 120, so `bun run format:check`
fails `build / agent-service` on both OSes and `Required Checks`. The format
check runs first, so CI has not run the tests yet. This is Prettier's own
output:
```suggestion
return {
response: "done",
messages: [],
usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 },
stopped: false,
};
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
