| author | |
| committer | |
| log | 7ac93e300ee17a4ee5e92192264fcfd438805b6f |
| tree | 101922130c793ab58ea213b3e03f65b095dd59bf |
| parent | c2c4185e175daea86f8fd6336fd8839a81cc616e |
| signature | Signed by PGP key B5690EEEBB952194 |
3 files changed, 151 insertions(+), 14 deletions(-)
apps/server/src/orchestration/decider.settled.test.ts+106| ... | ... | @@ -322,6 +322,112 @@ it.layer(NodeServices.layer)("settled thread decider", (it) => { |
| 322 | 322 | }), |
| 323 | 323 | ); |
| 324 | 324 | |
| 325 | it.effect("manual settlement dismisses async questions without starting a turn", () => | |
| 326 | Effect.gen(function* () { | |
| 327 | const question = (requestId: string): OrchestrationThread["activities"][number] => ({ | |
| 328 | id: EventId.make(requestId), | |
| 329 | kind: "user-input.requested", | |
| 330 | summary: "Question", | |
| 331 | tone: "approval", | |
| 332 | turnId: null, | |
| 333 | createdAt: "1969-12-31T00:00:00.000Z", | |
| 334 | payload: { requestId, responseMode: "message" }, | |
| 335 | }); | |
| 336 | const readModel = makeReadModel(null, null, makeSession("ready"), [ | |
| 337 | question("first"), | |
| 338 | question("second"), | |
| 339 | question("answered"), | |
| 340 | { | |
| 341 | ...question("answered"), | |
| 342 | id: EventId.make("answer"), | |
| 343 | createdAt: "1969-12-31T01:00:00.000Z", | |
| 344 | kind: "user-input.resolved", | |
| 345 | }, | |
| 346 | ]); | |
| 347 | const command = { | |
| 348 | type: "thread.settle" as const, | |
| 349 | commandId: CommandId.make("settle-async"), | |
| 350 | threadId: ThreadId.make("thread-1"), | |
| 351 | }; | |
| 352 | const result = yield* decideOrchestrationCommand({ command, readModel }); | |
| 353 | const events = Array.isArray(result) ? result : [result]; | |
| 354 | expect(events.map((event) => event.type)).toEqual([ | |
| 355 | "thread.settled", | |
| 356 | "thread.activity-appended", | |
| 357 | "thread.activity-appended", | |
| 358 | ]); | |
| 359 | expect(events.slice(1).map((event) => event.payload)).toEqual( | |
| 360 | ["first", "second"].map((requestId) => ({ | |
| 361 | threadId: command.threadId, | |
| 362 | activity: expect.objectContaining({ | |
| 363 | kind: "user-input.resolved", | |
| 364 | summary: "User input dismissed", | |
| 365 | payload: { requestId, responseMode: "message" }, | |
| 366 | }), | |
| 367 | })), | |
| 368 | ); | |
| 369 | let projected = readModel; | |
| 370 | for (const [index, event] of events.entries()) { | |
| 371 | projected = yield* projectEvent(projected, { ...event, sequence: index + 1 }); | |
| 372 | } | |
| 373 | expect(projected.threads[0]?.settledOverride).toBe("settled"); | |
| 374 | expect(projected.threads[0]?.messages).toEqual([]); | |
| 375 | const repeated = yield* decideOrchestrationCommand({ command, readModel: projected }); | |
| 376 | expect(repeated).toMatchObject({ type: "thread.settled" }); | |
| 377 | }), | |
| 378 | ); | |
| 379 | ||
| 380 | it.effect("async questions do not bypass automatic settlement or other blockers", () => | |
| 381 | Effect.gen(function* () { | |
| 382 | const question: OrchestrationThread["activities"][number] = { | |
| 383 | id: EventId.make("async-question"), | |
| 384 | kind: "user-input.requested", | |
| 385 | summary: "Question", | |
| 386 | tone: "approval", | |
| 387 | turnId: null, | |
| 388 | createdAt: NOW, | |
| 389 | payload: { requestId: "async-question", responseMode: "message" }, | |
| 390 | }; | |
| 391 | for (const blocker of ["auto", "running", "starting", "approval", "native"] as const) { | |
| 392 | const error = yield* decideOrchestrationCommand({ | |
| 393 | command: | |
| 394 | blocker === "auto" | |
| 395 | ? { | |
| 396 | type: "thread.auto-settle", | |
| 397 | commandId: CommandId.make(`settle-${blocker}`), | |
| 398 | threadId: ThreadId.make("thread-1"), | |
| 399 | snapshotSequence: 0, | |
| 400 | settledAt: NOW, | |
| 401 | } | |
| 402 | : { | |
| 403 | type: "thread.settle", | |
| 404 | commandId: CommandId.make(`settle-${blocker}`), | |
| 405 | threadId: ThreadId.make("thread-1"), | |
| 406 | }, | |
| 407 | readModel: makeReadModel( | |
| 408 | null, | |
| 409 | null, | |
| 410 | makeSession(blocker === "running" || blocker === "starting" ? blocker : "ready"), | |
| 411 | [ | |
| 412 | question, | |
| 413 | ...(blocker === "approval" || blocker === "native" | |
| 414 | ? [ | |
| 415 | { | |
| 416 | ...question, | |
| 417 | id: EventId.make("blocking-request"), | |
| 418 | kind: blocker === "approval" ? "approval.requested" : "user-input.requested", | |
| 419 | payload: { requestId: "blocking-request" }, | |
| 420 | }, | |
| 421 | ] | |
| 422 | : []), | |
| 423 | ], | |
| 424 | ), | |
| 425 | }).pipe(Effect.flip); | |
| 426 | expect(error).toMatchObject({ _tag: "OrchestrationThreadSettleBlockedError" }); | |
| 427 | } | |
| 428 | }), | |
| 429 | ); | |
| 430 | ||
| 325 | 431 | it.effect("clears an open request when its respond failure marks it stale", () => |
| 326 | 432 | Effect.gen(function* () { |
| 327 | 433 | const activity = ( |
apps/server/src/orchestration/decider.ts+43-14| ... | ... | @@ -68,10 +68,8 @@ function isStaleRequestFailureDetail(payload: Record<string, unknown> | null): b |
| 68 | 68 | // Scans the read model's activities, which the projector caps at the most |
| 69 | 69 | // recent 500 plus pending async questions. Async questions remain actionable |
| 70 | 70 | // while the agent works, so they must not expire with the activity window. |
| 71 | function hasOpenBlockingRequest(thread: { | |
| 72 | readonly activities: ReadonlyArray<{ readonly kind: string; readonly payload: unknown }>; | |
| 73 | }): boolean { | |
| 74 | const openRequestIds = new Set<string>(); | |
| 71 | function openRequests(thread: Pick<OrchestrationThread, "activities">) { | |
| 72 | const requests = new Map<string, OrchestrationThreadActivity>(); | |
| 75 | 73 | for (const activity of thread.activities) { |
| 76 | 74 | const payload = |
| 77 | 75 | typeof activity.payload === "object" && activity.payload !== null |
| ... | ... | @@ -80,18 +78,18 @@ function hasOpenBlockingRequest(thread: { |
| 80 | 78 | const requestId = typeof payload?.requestId === "string" ? payload.requestId : null; |
| 81 | 79 | if (requestId === null) continue; |
| 82 | 80 | if (activity.kind === "approval.requested" || activity.kind === "user-input.requested") { |
| 83 | openRequestIds.add(requestId); | |
| 81 | requests.set(requestId, activity); | |
| 84 | 82 | } else if (activity.kind === "approval.resolved" || activity.kind === "user-input.resolved") { |
| 85 | openRequestIds.delete(requestId); | |
| 83 | requests.delete(requestId); | |
| 86 | 84 | } else if ( |
| 87 | 85 | (activity.kind === "provider.approval.respond.failed" || |
| 88 | 86 | activity.kind === "provider.user-input.respond.failed") && |
| 89 | 87 | isStaleRequestFailureDetail(payload) |
| 90 | 88 | ) { |
| 91 | openRequestIds.delete(requestId); | |
| 89 | requests.delete(requestId); | |
| 92 | 90 | } |
| 93 | 91 | } |
| 94 | return openRequestIds.size > 0; | |
| 92 | return requests; | |
| 95 | 93 | } |
| 96 | 94 | |
| 97 | 95 | /** Apply the shared shell-level rule to the detailed command read model. */ |
| ... | ... | @@ -456,10 +454,18 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" |
| 456 | 454 | if (thread.session?.status === "starting" || thread.session?.status === "running") { |
| 457 | 455 | return yield* new OrchestrationThreadSettleBlockedError({ threadId: command.threadId }); |
| 458 | 456 | } |
| 459 | // Pending approval / user-input requests are blocked-on-you work: a | |
| 460 | // raced or stale client must not park them behind a settled override | |
| 461 | // that would surface only after the request resolves. | |
| 462 | if (hasOpenBlockingRequest(thread)) { | |
| 457 | const pendingRequests = openRequests(thread); | |
| 458 | // Manual settlement dismisses async questions without answering them. | |
| 459 | // Native callbacks and approvals still need a response or interruption. | |
| 460 | if ( | |
| 461 | Array.from(pendingRequests.values()).some( | |
| 462 | (activity) => | |
| 463 | command.type === "thread.auto-settle" || | |
| 464 | activity.kind !== "user-input.requested" || | |
| 465 | !Predicate.isObject(activity.payload) || | |
| 466 | activity.payload.responseMode !== "message", | |
| 467 | ) | |
| 468 | ) { | |
| 463 | 469 | return yield* new OrchestrationThreadSettleBlockedError({ threadId: command.threadId }); |
| 464 | 470 | } |
| 465 | 471 | const occurredAt = yield* nowIso; |
| ... | ... | @@ -495,6 +501,29 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" |
| 495 | 501 | // Settling is "I'm done with this": clear states that would keep the |
| 496 | 502 | // row pinned or snoozed instead of showing the new settled state. |
| 497 | 503 | const companionEvents: Array<Omit<OrchestrationEvent, "sequence">> = []; |
| 504 | for (const [requestId, request] of pendingRequests) { | |
| 505 | companionEvents.push({ | |
| 506 | ...(yield* withEventBase({ | |
| 507 | aggregateKind: "thread", | |
| 508 | aggregateId: command.threadId, | |
| 509 | occurredAt, | |
| 510 | commandId: command.commandId, | |
| 511 | })), | |
| 512 | type: "thread.activity-appended", | |
| 513 | payload: { | |
| 514 | threadId: command.threadId, | |
| 515 | activity: { | |
| 516 | id: EventId.make(`settle:${command.commandId}:${requestId}`), | |
| 517 | kind: "user-input.resolved", | |
| 518 | summary: "User input dismissed", | |
| 519 | tone: "info", | |
| 520 | turnId: request.turnId, | |
| 521 | createdAt: occurredAt, | |
| 522 | payload: { requestId, responseMode: "message" }, | |
| 523 | }, | |
| 524 | }, | |
| 525 | }); | |
| 526 | } | |
| 498 | 527 | if (thread.pinnedAt != null) { |
| 499 | 528 | companionEvents.push({ |
| 500 | 529 | ...(yield* withEventBase({ |
| ... | ... | @@ -581,7 +610,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" |
| 581 | 610 | // user-input request is the agent waiting on the user, and hiding it |
| 582 | 611 | // defeats the request. (A running session IS snoozable — snooze only |
| 583 | 612 | // affects visibility, never the agent.) |
| 584 | if (hasOpenBlockingRequest(thread)) { | |
| 613 | if (openRequests(thread).size > 0) { | |
| 585 | 614 | return yield* Effect.fail( |
| 586 | 615 | new OrchestrationCommandInvariantError({ |
| 587 | 616 | commandType: command.type, |
| ... | ... | @@ -1453,7 +1482,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" |
| 1453 | 1482 | thread.messages.length > 0 || |
| 1454 | 1483 | thread.latestTurn !== null || |
| 1455 | 1484 | thread.session !== null || |
| 1456 | hasOpenBlockingRequest(thread) | |
| 1485 | openRequests(thread).size > 0 | |
| 1457 | 1486 | ) { |
| 1458 | 1487 | return yield* new OrchestrationCommandInvariantError({ |
| 1459 | 1488 | commandType: command.type, |
docs/user/thread-sidebar.md+2| ... | ... | @@ -63,6 +63,8 @@ their default order until the server is updated. |
| 63 | 63 | Choose **Settle thread** from its menu to move finished work out of the active list |
| 64 | 64 | without deleting the conversation. **Un-settle thread** restores it to active work |
| 65 | 65 | and prevents automatic settlement until new activity resumes the usual rules. |
| 66 | Manually settling an idle thread dismisses unanswered async questions without | |
| 67 | sending an answer or restarting the agent. | |
| 66 | 68 | |
| 67 | 69 | By default, environments settle inactive threads after three days and settle |
| 68 | 70 | threads whose pull request merged. A closed pull request can also settle an idle |