Skip to content

Commit d180cab

Browse files
SaladDaySaladDay
authored
Remove handling of legacy Session deletion markers (#799)
1 parent 85c9496 commit d180cab

14 files changed

Lines changed: 132 additions & 399 deletions

‎services/core/internal/db/queries/scheduling.sql‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ SELECT pg_try_advisory_lock(706172736172::bigint)::boolean;
55
SELECT t.id, t.session_id, s.tenant_id, t.status
66
FROM turns t JOIN sessions s ON s.id = t.session_id
77
WHERE t.status = ANY(sqlc.arg(statuses)::text[]) AND t.id > sqlc.arg(after_id)::uuid
8-
AND (s.deleted_at IS NULL OR t.status <> 'queued')
98
AND (NOT sqlc.arg(connected_only)::boolean OR EXISTS (
109
SELECT 1 FROM devices d WHERE d.tenant_id = s.tenant_id AND d.revoked_at IS NULL
1110
AND d.id = ANY(sqlc.arg(connected_devices)::uuid[])

‎services/core/internal/db/sqlc/scheduling.sql.go‎

Lines changed: 0 additions & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎services/core/internal/sessions/turn.go‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -108,8 +108,7 @@ type TurnReader interface {
108108
// cursor Turn is ErrNotFound.
109109
ListTurns(ctx context.Context, tenant, session, cursor string, limit int, ascending bool) (TurnPage, error)
110110
// ListExecutionWork lists, in ID order after the Turn after names, up to
111-
// 100 root Turns in statuses across tenants; the queued Turns of a publicly
112-
// deleted Session are not work. With connectedDevices not nil, it lists
113-
// only the Turns of tenants that own one of those devices, unrevoked.
111+
// 100 root Turns in statuses across tenants. With connectedDevices not nil, it
112+
// lists only the Turns of tenants that own one of those devices, unrevoked.
114113
ListExecutionWork(ctx context.Context, after string, statuses, connectedDevices []string) ([]ExecutionWork, error)
115114
}

‎services/core/tests/integration/environment_admission_test.go‎

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -164,7 +164,7 @@ func TestEnvironmentAdmissionWaitsForPreparedClaimAndRetainsRetry(t *testing.T)
164164
}
165165

166166
func TestEnvironmentAdmissionSettlementDoesNotCreateTurn(t *testing.T) {
167-
for _, name := range []string{"expired", "cancelled", "deleted", "disconnected", "ownership_lost"} {
167+
for _, name := range []string{"expired", "cancelled", "disconnected", "ownership_lost"} {
168168
t.Run(name, func(t *testing.T) {
169169
h, worker := newEnvironmentAdmission(t)
170170
ctx, cancel := context.WithCancel(t.Context())
@@ -183,14 +183,6 @@ func TestEnvironmentAdmissionSettlementDoesNotCreateTurn(t *testing.T) {
183183
if _, err := cancelEnvironmentInput(t.Context(), h.s, h.tenant, h.session.ID, pending.ID); err != nil {
184184
t.Fatal(err)
185185
}
186-
case "deleted":
187-
expected = sessions.ErrNotFound
188-
if err := sessionService(t, h.s).DeleteSession(t.Context(), sessions.DeleteSessionCommand{TenantID: h.tenant, SessionID: h.session.ID}); !errors.Is(err, sessions.ErrNotIdle) {
189-
t.Fatal("pending input deleted", err)
190-
}
191-
if err := h.s.commitLegacyDeletion(t.Context(), h.tenant, h.session.ID); err != nil {
192-
t.Fatal(err)
193-
}
194186
case "disconnected":
195187
expected = context.Canceled
196188
cancel()
@@ -216,9 +208,6 @@ func TestEnvironmentAdmissionSettlementDoesNotCreateTurn(t *testing.T) {
216208
if turns != 0 || inputs != 0 {
217209
t.Fatal("settlement created work", turns, inputs)
218210
}
219-
if name == "deleted" {
220-
return
221-
}
222211
retained, err := sessionAdapter(h.s).GetEnvironmentInputReservation(t.Context(), h.tenant, h.session.ID, pending.ID)
223212
if err != nil {
224213
t.Fatal(err)

‎services/core/tests/integration/environment_claim_worker_test.go‎

Lines changed: 45 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -13,66 +13,50 @@ import (
1313
)
1414

1515
func TestWorkerReconcilesEnvironmentPromotionBeforeStart(t *testing.T) {
16-
for _, deleted := range []bool{false, true} {
17-
t.Run(map[bool]string{false: "unbound", true: "deleted"}[deleted], func(t *testing.T) {
18-
s, _ := testStore(t)
19-
tenant, pending := newEnvironmentExpiryReservation(t, s)
20-
owner := executionOwner(t, s)
21-
got, err := owner.Sessions.PromoteEnvironmentInput(t.Context(), tenant, pending.SessionID, pending.ID)
22-
if err != nil || len(got.Receipts) != 1 || got.Receipts[0].Replayed {
23-
t.Fatal(got, err)
24-
}
25-
turnID := got.Receipts[0].TurnID
26-
if deleted {
27-
if err := sessionService(t, s).DeleteSession(t.Context(), sessions.DeleteSessionCommand{TenantID: tenant, SessionID: pending.SessionID}); !errors.Is(err, sessions.ErrNotIdle) {
28-
t.Fatal("claimed Session deleted", err)
29-
}
30-
if err := s.commitLegacyDeletion(t.Context(), tenant, pending.SessionID); err != nil {
31-
t.Fatal(err)
32-
}
33-
}
34-
turn, err := sessionAdapter(s).GetTurn(t.Context(), tenant, pending.SessionID, turnID)
35-
if err != nil || turn.Status != sessions.TurnInProgress || (deleted && turn.CancelRequestedAt.IsZero()) {
36-
t.Fatal("promotion did not retain the active claim", turn, err)
37-
}
38-
// Simulate owner loss after commit, without sending any daemon Start.
39-
awaitRelease := pgtest.ObserveExecutionLeaseRelease(t, s.pool)
40-
if err := owner.Lease.Close(t.Context()); err != nil {
41-
t.Fatal(err)
42-
}
43-
awaitRelease()
44-
restarted := startWorker(t, t.Context(), s, &execution.Dispatcher{Registry: runtimegateway.NewRegistry()})
45-
stopped, cancel := context.WithCancel(t.Context())
46-
cancel()
47-
awaitRelease = pgtest.ObserveExecutionLeaseRelease(t, s.pool)
48-
if err := restarted.Run(stopped); !errors.Is(err, context.Canceled) {
49-
t.Fatal(err)
50-
}
51-
awaitRelease()
52-
turn, err = sessionAdapter(s).GetTurn(t.Context(), tenant, pending.SessionID, turnID)
53-
var outcome struct {
54-
ErrorCode string `json:"error_code"`
55-
}
56-
if err != nil || turn.Status != sessions.TurnFailed || json.Unmarshal(turn.Outcome, &outcome) != nil || outcome.ErrorCode != "execution_interrupted" {
57-
t.Fatal("restart failed to settle the original claim", turn, err)
58-
}
59-
var turns, inputs, queued int
60-
err = s.pool.QueryRow(t.Context(), `SELECT
61-
(SELECT count(*) FROM turns WHERE session_id=$1),
62-
(SELECT count(*) FROM turn_inputs WHERE session_id=$1),
63-
(SELECT count(*) FROM turns WHERE session_id=$1 AND status='queued')`, pending.SessionID).Scan(&turns, &inputs, &queued)
64-
if err != nil || turns != 1 || inputs != 1 || queued != 0 {
65-
t.Fatal("restart duplicated or requeued prepared work", turns, inputs, queued, err)
66-
}
67-
successor := executionOwner(t, s).Sessions
68-
retry, err := successor.PromoteEnvironmentInput(t.Context(), tenant, pending.SessionID, pending.ID)
69-
if deleted {
70-
if !errors.Is(err, sessions.ErrNotFound) {
71-
t.Fatal("deleted reservation was exposed", err)
72-
}
73-
} else if err != nil || len(retry.Receipts) != 1 || !retry.Receipts[0].Replayed || retry.Receipts[0].Sequence != got.Receipts[0].Sequence || retry.Receipts[0].TurnID != turnID {
74-
t.Fatal("recovery retry authorized another start", retry, err)
75-
}
76-
})
16+
s, _ := testStore(t)
17+
tenant, pending := newEnvironmentExpiryReservation(t, s)
18+
owner := executionOwner(t, s)
19+
got, err := owner.Sessions.PromoteEnvironmentInput(t.Context(), tenant, pending.SessionID, pending.ID)
20+
if err != nil || len(got.Receipts) != 1 || got.Receipts[0].Replayed {
21+
t.Fatal(got, err)
22+
}
23+
turnID := got.Receipts[0].TurnID
24+
turn, err := sessionAdapter(s).GetTurn(t.Context(), tenant, pending.SessionID, turnID)
25+
if err != nil || turn.Status != sessions.TurnInProgress {
26+
t.Fatal("promotion did not retain the active claim", turn, err)
27+
}
28+
// Simulate owner loss after commit, without sending any daemon Start.
29+
awaitRelease := pgtest.ObserveExecutionLeaseRelease(t, s.pool)
30+
if err := owner.Lease.Close(t.Context()); err != nil {
31+
t.Fatal(err)
32+
}
33+
awaitRelease()
34+
restarted := startWorker(t, t.Context(), s, &execution.Dispatcher{Registry: runtimegateway.NewRegistry()})
35+
stopped, cancel := context.WithCancel(t.Context())
36+
cancel()
37+
awaitRelease = pgtest.ObserveExecutionLeaseRelease(t, s.pool)
38+
if err := restarted.Run(stopped); !errors.Is(err, context.Canceled) {
39+
t.Fatal(err)
40+
}
41+
awaitRelease()
42+
turn, err = sessionAdapter(s).GetTurn(t.Context(), tenant, pending.SessionID, turnID)
43+
var outcome struct {
44+
ErrorCode string `json:"error_code"`
45+
}
46+
if err != nil || turn.Status != sessions.TurnFailed || json.Unmarshal(turn.Outcome, &outcome) != nil || outcome.ErrorCode != "execution_interrupted" {
47+
t.Fatal("restart failed to settle the original claim", turn, err)
48+
}
49+
var turns, inputs, queued int
50+
err = s.pool.QueryRow(t.Context(), `SELECT
51+
(SELECT count(*) FROM turns WHERE session_id=$1),
52+
(SELECT count(*) FROM turn_inputs WHERE session_id=$1),
53+
(SELECT count(*) FROM turns WHERE session_id=$1 AND status='queued')`, pending.SessionID).Scan(&turns, &inputs, &queued)
54+
if err != nil || turns != 1 || inputs != 1 || queued != 0 {
55+
t.Fatal("restart duplicated or requeued prepared work", turns, inputs, queued, err)
56+
}
57+
successor := executionOwner(t, s).Sessions
58+
retry, err := successor.PromoteEnvironmentInput(t.Context(), tenant, pending.SessionID, pending.ID)
59+
if err != nil || len(retry.Receipts) != 1 || !retry.Receipts[0].Replayed || retry.Receipts[0].Sequence != got.Receipts[0].Sequence || retry.Receipts[0].TurnID != turnID {
60+
t.Fatal("recovery retry authorized another start", retry, err)
7761
}
7862
}

‎services/core/tests/integration/environment_input_activity_test.go‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -245,13 +245,4 @@ func TestEnvironmentInputActivityRecoversWaitingActionAndHidesDeletion(t *testin
245245
if err := sessionService(t, s).DeleteSession(t.Context(), sessions.DeleteSessionCommand{TenantID: tenant, SessionID: session.ID}); !errors.Is(err, sessions.ErrNotIdle) {
246246
t.Fatal("waiting input deleted", err)
247247
}
248-
if err := s.commitLegacyDeletion(t.Context(), tenant, session.ID); err != nil {
249-
t.Fatal(err)
250-
}
251-
if _, err := sessionAdapter(s).GetSession(t.Context(), tenant, session.ID); !errors.Is(err, sessions.ErrNotFound) {
252-
t.Fatal("deleted activity remained visible", err)
253-
}
254-
if _, err := sessionAdapter(s).ListSessionEvents(t.Context(), tenant, session.ID, 0); !errors.Is(err, sessions.ErrNotFound) {
255-
t.Fatal("deleted activity events remained visible", err)
256-
}
257248
}

‎services/core/tests/integration/environment_input_settlement_test.go‎

Lines changed: 0 additions & 67 deletions
This file was deleted.

‎services/core/tests/integration/environment_work_test.go‎

Lines changed: 1 addition & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,8 @@
11
package integration
22

33
import (
4-
"errors"
54
"testing"
65

7-
"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions"
86
"github.com/google/uuid"
97
)
108

@@ -16,7 +14,7 @@ func TestEnvironmentInputWorkFiltersAndPagesDevices(t *testing.T) {
1614
pending := workerEnvironmentReservation(t, h)
1715
wanted[pending.ID] = pending.SessionID
1816
}
19-
for _, state := range []string{"expired", "cancelled", "deleted", "unbound", "revoked", "disconnected"} {
17+
for _, state := range []string{"expired", "cancelled", "unbound", "revoked", "disconnected"} {
2018
pending := workerEnvironmentReservation(t, h)
2119
switch state {
2220
case "expired":
@@ -25,13 +23,6 @@ func TestEnvironmentInputWorkFiltersAndPagesDevices(t *testing.T) {
2523
if _, err := cancelEnvironmentInput(t.Context(), h.s, h.tenant, pending.SessionID, pending.ID); err != nil {
2624
t.Fatal(err)
2725
}
28-
case "deleted":
29-
if err := sessionService(t, h.s).DeleteSession(t.Context(), sessions.DeleteSessionCommand{TenantID: h.tenant, SessionID: pending.SessionID}); !errors.Is(err, sessions.ErrNotIdle) {
30-
t.Fatal("pending input deleted", err)
31-
}
32-
if err := h.s.commitLegacyDeletion(t.Context(), h.tenant, pending.SessionID); err != nil {
33-
t.Fatal(err)
34-
}
3526
case "unbound":
3627
wanted[pending.ID] = pending.SessionID
3728
if _, err := pool.Exec(t.Context(), "DELETE FROM session_devices WHERE session_id=$1", pending.SessionID); err != nil {

‎services/core/tests/integration/prepared_dispatch_failure_test.go‎

Lines changed: 1 addition & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ import (
1414
)
1515

1616
func TestPreparedDispatchSettlesOnlyReadyInput(t *testing.T) {
17-
for _, action := range []string{"cancel", "expire", "delete", "prepare-failure", "disconnect"} {
17+
for _, action := range []string{"cancel", "expire", "prepare-failure", "disconnect"} {
1818
t.Run(action, func(t *testing.T) {
1919
h, pending := preparedDispatchHarness(t)
2020
_, pool := testStore(t)
@@ -30,13 +30,6 @@ func TestPreparedDispatchSettlesOnlyReadyInput(t *testing.T) {
3030
if _, err := pool.Exec(t.Context(), "UPDATE environment_input_reservations SET deadline=clock_timestamp()-interval '1 second' WHERE id=$1", pending.ID); err != nil {
3131
t.Fatal(err)
3232
}
33-
case "delete":
34-
if err := sessionService(t, h.s).DeleteSession(t.Context(), sessions.DeleteSessionCommand{TenantID: h.tenant, SessionID: h.session.ID}); !errors.Is(err, sessions.ErrNotIdle) {
35-
t.Fatal("pending input deleted", err)
36-
}
37-
if err := h.s.commitLegacyDeletion(t.Context(), h.tenant, h.session.ID); err != nil {
38-
t.Fatal(err)
39-
}
4033
case "prepare-failure":
4134
h.write(frame.ID, proto.TypePreparationStatus, proto.PreparationStatusPayload{Handle: handle, Revision: 2, State: "failed", ErrorCode: "preparation_failed"})
4235
case "disconnect":
@@ -57,9 +50,6 @@ func TestPreparedDispatchSettlesOnlyReadyInput(t *testing.T) {
5750
} else if got.err == nil {
5851
t.Fatal("preparation failure was hidden")
5952
}
60-
if action == "delete" && !errors.Is(got.err, sessions.ErrNotFound) {
61-
t.Fatal("deleted reservation remained accessible", got.err)
62-
}
6353
if action != "disconnect" {
6454
assertPreparationReleased(t, h, frame.ID, handle)
6555
}

‎services/core/tests/integration/session_artifacts_test.go‎

Lines changed: 25 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -210,47 +210,31 @@ func TestSessionArtifactsDiscardTerminalPrivateCapture(t *testing.T) {
210210
}
211211
}
212212

213-
func TestSessionArtifactTransferDoesNotBlockDeletionOrCancellation(t *testing.T) {
214-
for _, operation := range []string{"delete", "cancel"} {
215-
t.Run(operation, func(t *testing.T) {
216-
s, _ := configuredStore(t)
217-
pool := s.pool
218-
tenant, session, environment, turn := artifactTurn(t, s, "openai_hosted")
219-
before := largeObjectCount(t, pool)
220-
reader, writer := io.Pipe()
221-
defer reader.Close()
222-
defer writer.Close()
223-
result := make(chan error, 1)
224-
go func() { result <- stageTurnArtifacts(t.Context(), s, tenant, session, turn, environment, reader) }()
225-
// A complete TAR arrives, but transport has not acknowledged success yet.
226-
if _, err := writer.Write(artifactArchive(t, map[string][]byte{"outputs/a": []byte("partial")})); err != nil {
227-
t.Fatal(err)
228-
}
229-
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
230-
defer cancel()
231-
want := sessions.ErrNotFound
232-
if operation == "delete" {
233-
// The idle-only decision itself is not blocked by the transfer.
234-
if err := sessionService(t, s).DeleteSession(ctx, sessions.DeleteSessionCommand{TenantID: tenant, SessionID: session}); !errors.Is(err, sessions.ErrNotIdle) {
235-
t.Fatalf("transfer blocked or bypassed the deletion rule: %v", err)
236-
}
237-
if err := s.commitLegacyDeletion(ctx, tenant, session); err != nil {
238-
t.Fatalf("transfer blocked deletion: %v", err)
239-
}
240-
} else {
241-
if _, err := requestCancel(ctx, s, tenant, session, "cancel-capture"); err != nil {
242-
t.Fatalf("transfer blocked cancellation: %v", err)
243-
}
244-
want = sessions.ErrTurnConflict
245-
}
246-
writer.Close()
247-
if err := <-result; !errors.Is(err, want) {
248-
t.Fatalf("late publication after %s: %v", operation, err)
249-
}
250-
if count := largeObjectCount(t, pool); count != before {
251-
t.Fatalf("late capture leaked objects: %d -> %d", before, count)
252-
}
253-
})
213+
func TestSessionArtifactTransferDoesNotBlockCancellation(t *testing.T) {
214+
s, _ := configuredStore(t)
215+
pool := s.pool
216+
tenant, session, environment, turn := artifactTurn(t, s, "openai_hosted")
217+
before := largeObjectCount(t, pool)
218+
reader, writer := io.Pipe()
219+
defer reader.Close()
220+
defer writer.Close()
221+
result := make(chan error, 1)
222+
go func() { result <- stageTurnArtifacts(t.Context(), s, tenant, session, turn, environment, reader) }()
223+
// A complete TAR arrives, but transport has not acknowledged success yet.
224+
if _, err := writer.Write(artifactArchive(t, map[string][]byte{"outputs/a": []byte("partial")})); err != nil {
225+
t.Fatal(err)
226+
}
227+
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
228+
defer cancel()
229+
if _, err := requestCancel(ctx, s, tenant, session, "cancel-capture"); err != nil {
230+
t.Fatalf("transfer blocked cancellation: %v", err)
231+
}
232+
writer.Close()
233+
if err := <-result; !errors.Is(err, sessions.ErrTurnConflict) {
234+
t.Fatalf("late publication after cancellation: %v", err)
235+
}
236+
if count := largeObjectCount(t, pool); count != before {
237+
t.Fatalf("late capture leaked objects: %d -> %d", before, count)
254238
}
255239
}
256240

0 commit comments

Comments
 (0)