File size: 4,648 Bytes
4c76b0d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
import { afterEach, describe, expect, it } from "vitest";
import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../../infra/kysely-sync.js";
import type { DB as OpenClawAgentKyselyDatabase } from "../../../state/openclaw-agent-db.generated.js";
import {
  closeOpenClawAgentDatabasesForTest,
  runOpenClawAgentWriteTransaction,
} from "../../../state/openclaw-agent-db.js";
import { withTestDir } from "../../../test-helpers/temp-dir.js";
import { recordAcpParentStreamEvents } from "./acp-parent-stream-store.sqlite.js";
import { listAcpParentStreamEventsForTest } from "./acp-parent-stream-store.sqlite.test-support.js";

describe("ACP parent stream SQLite store", () => {
  afterEach(() => {
    closeOpenClawAgentDatabasesForTest();
  });

  it("orders run events and removes them with the child session", async () => {
    await withTestDir({ prefix: "openclaw-acp-parent-stream-" }, async (stateDir) => {
      const options = {
        agentId: "codex",
        env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
      };
      runOpenClawAgentWriteTransaction((database) => {
        const db = getNodeSqliteKysely<
          Pick<OpenClawAgentKyselyDatabase, "session_nodes" | "session_windows">
        >(database.db);
        executeSqliteQuerySync(
          database.db,
          db.insertInto("session_nodes").values({
            session_key: "agent:codex:acp:child",
            current_session_id: "session-1",
            entry_json: "{}",
            updated_at: 1,
          }),
        );
        executeSqliteQuerySync(
          database.db,
          db.insertInto("session_windows").values({
            session_id: "session-1",
            session_key: "agent:codex:acp:child",
            session_scope: "conversation",
            created_at: 1,
            updated_at: 1,
          }),
        );
      }, options);

      recordAcpParentStreamEvents({
        ...options,
        sessionId: "session-1",
        runId: "run-1",
        events: [
          { createdAt: 10, event: { kind: "assistant_delta", delta: "one" } },
          { createdAt: 11, event: { kind: "lifecycle", phase: "end" } },
        ],
      });

      expect(
        listAcpParentStreamEventsForTest({ ...options, sessionId: "session-1", runId: "run-1" }),
      ).toEqual([
        { kind: "assistant_delta", delta: "one" },
        { kind: "lifecycle", phase: "end" },
      ]);

      runOpenClawAgentWriteTransaction((database) => {
        const db = getNodeSqliteKysely<Pick<OpenClawAgentKyselyDatabase, "session_windows">>(
          database.db,
        );
        executeSqliteQuerySync(
          database.db,
          db.deleteFrom("session_windows").where("session_id", "=", "session-1"),
        );
      }, options);
      expect(
        listAcpParentStreamEventsForTest({ ...options, sessionId: "session-1", runId: "run-1" }),
      ).toEqual([]);
    });
  });

  it("drops unserializable events without blocking later diagnostics", async () => {
    await withTestDir({ prefix: "openclaw-acp-parent-stream-invalid-" }, async (stateDir) => {
      const options = {
        agentId: "codex",
        env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
      };
      runOpenClawAgentWriteTransaction((database) => {
        const db = getNodeSqliteKysely<
          Pick<OpenClawAgentKyselyDatabase, "session_nodes" | "session_windows">
        >(database.db);
        executeSqliteQuerySync(
          database.db,
          db.insertInto("session_nodes").values({
            session_key: "agent:codex:acp:invalid",
            current_session_id: "session-1",
            entry_json: "{}",
            updated_at: 1,
          }),
        );
        executeSqliteQuerySync(
          database.db,
          db.insertInto("session_windows").values({
            session_id: "session-1",
            session_key: "agent:codex:acp:invalid",
            session_scope: "conversation",
            created_at: 1,
            updated_at: 1,
          }),
        );
      }, options);
      const circular: Record<string, unknown> = { kind: "circular" };
      circular.self = circular;

      recordAcpParentStreamEvents({
        ...options,
        sessionId: "session-1",
        runId: "run-1",
        events: [
          { createdAt: 10, event: { toJSON: () => undefined } },
          { createdAt: 11, event: circular },
          { createdAt: 12, event: { kind: "lifecycle", phase: "end" } },
        ],
      });

      expect(
        listAcpParentStreamEventsForTest({ ...options, sessionId: "session-1", runId: "run-1" }),
      ).toEqual([{ kind: "lifecycle", phase: "end" }]);
    });
  });
});