Durable 持久执行

import io.agentscope.core.ReActAgent;
import io.agentscope.durable.DurableRuntime;
import io.agentscope.durable.DurableTypes.SessionKey;
import io.agentscope.durable.agentscope.AgentScopeAdapter;
import io.agentscope.extensions.durable.jdbc.LocalBackend;
import java.nio.file.Path;
import java.util.Map;

var backend = LocalBackend.open(Path.of("./durable"));
var adapter = AgentScopeAdapter.forReAct(binding ->
        ReActAgent.builder().name("assistant").model(model), Map.of());
var runtime = DurableRuntime.builder().backend(backend)
        .harness("assistant", "definition-v1", adapter).build();
var session = new SessionKey("application", "conversation-1");
runtime.createSession(session, "assistant");
var receipt = runtime.getSession(session).submit("request-1", "Describe this workspace");
runtime.start();

var conversation = runtime.getSession(new SessionKey("application", "conversation-1"));
var thread = conversation.getThread("main");
var turn = thread.getTurn(savedTurnId);
var current = turn.status();
var history = thread.history(null, 100);

var investigation = conversation.submit("investigate-1", "调查导出超时的原因");
var summary = conversation.submit("summary-1", "总结调查结果");
investigation.steer("constraint-1", "重点检查数据库访问，不要修改 API");
summary.cancel("cancel-summary-1");

import io.agentscope.durable.DurableTypes.TurnStatus;

var paused = turn.status();
if (paused.status() == TurnStatus.PAUSED) {
    turn.resume("resume-1", paused.revision());
}

var research = conversation.createThread("research", "assistant");
research.submit("research-1", "查找相关故障案例");
// 后续请求复用这个 Agent 的上下文。
conversation.getThread("research").submit("research-2", "比较这些案例的处理方案");

var handle = runtime.getSession(new SessionKey("application", "conversation-1"));
var turn = handle.submit("request-2", "继续研究");
var current = turn.status();
var history = handle.history(null, 100);
// 后续请求凭 SessionKey、threadId 和 turn.turnId() 获取原任务。

import io.agentscope.durable.codec.agentscope.AgentScopeCheckpointCodec;
import io.agentscope.durable.codec.agentscope.AgentScopeCheckpoints;

var api = DurableRuntime.builder().backend(backend)
        .definition("assistant", "definition-v1", AgentScopeCheckpointCodec.STATE_VERSION)
        .checkpoints("assistant", AgentScopeCheckpoints::restore)
        .build();
// API-only：不调用 start；这个 Runtime 不认领任务。

import io.agentscope.extensions.durable.http.DurableHttp;
import java.net.URI;

var web = DurableRuntime.builder()
        .transport(new DurableHttp(URI.create("http://127.0.0.1:8080/api")))
        .build();
var sameSession = web.getSession(new SessionKey("application", "conversation-1"));

import io.agentscope.durable.DurableMessages;
import java.util.List;

var input = new DurableMessages.Input("1", List.of(
        new DurableMessages.Text("描述这张图片"),
        new DurableMessages.Image(
                new DurableMessages.Url("https://example.org/image.png", "image/png"))));
var receipt = runtime.getSession(session).submit("picture-1", input);

import io.agentscope.durable.EventStreamOptions;
import io.agentscope.durable.EventStreamOptions.EventDeltaType;

var options = EventStreamOptions.builder()
        .after(savedCursor)
        .deltas(EventDeltaType.MESSAGE)
        .build();

try (var stream = session.streamEvents(options)) {
    stream.forEach(event -> render(event));
}

var worker = WorkerRuntime.builder()
        .recordStore(handsRecords)
        .executor("hands", toolExecutor)
        .build();

var executor = RocketMqRemoteExecutor.builder()
        .clientConfiguration(clientConfiguration)
        .build();
var runtime = DurableRuntime.builder().backend(backend)
        .harness("assistant", "v1", adapter)
        .executor("hands", executor).build();
runtime.start();

var worker = RocketMqWorker.builder()
        .executor("hands", toolExecutor).recordStore(handsRecords)
        .clientConfiguration(clientConfiguration)
        .authorize(command -> policy.check(command))
        .parallelism(4).build();
worker.start();

import io.agentscope.extensions.durable.jdbc.PostgresBackend;

// 部署阶段使用有 DDL 权限的身份初始化，先于应用启动执行：
PostgresBackend.initializeSchema(dataSource);

// 常规启动检查已有表结构，默认不执行 DDL。
var backend = PostgresBackend.builder()
        .dataSource(dataSource)
        .notifications(rocketMqWakeups) // 可选，默认使用进程内提示。
        .build();

import io.agentscope.durable.DurableTypes.RecoveryPolicy;

var runtime = DurableRuntime.builder()
        .backend(backend)
        .harness("agent", "v1", adapter)
        .recoveryPolicy(new RecoveryPolicy(3, 1000, 30000))
        .build();

import io.agentscope.durable.EventStreamOptions;
import io.agentscope.durable.EventProjection;
import io.agentscope.durable.SessionStreamEvent;

var projection = new EventProjection(session);
var subscription = runtime.getSession(session).subscribeEvents(
        EventStreamOptions.defaults(),
        event -> {
            if (event instanceof SessionStreamEvent.SnapshotReset reset) {
                projection.reset(reset.snapshot());
            } else if (event instanceof SessionStreamEvent.Committed committed) {
                projection.accept(committed.event());
            }
        });
// 无需编写轮询循环或配置观察线程。
// 连接存在期间保留订阅，观察者断开时调用 subscription.close()。

import io.agentscope.durable.DurableTypes.EventFilter;

var filter = new EventFilter("main", turnId, runId);
var scopedProjection = new EventProjection(session, filter);
var scoped = runtime.getSession(session)
        .subscribeEvents(filter, EventStreamOptions.defaults(), event -> apply(event));
// 从已有游标继续：EventStreamOptions.builder().after(savedCursor).build()

var local = new LocalSandboxClient(Path.of("./data/sandboxes"));
var workspace = new WorkspaceSpec();
workspace.getEntries().put("README.txt", new FileEntry("Durable workspace"));
var environment = new SandboxBindings.Binding(
        "demo", "workspace-1", "v1", DurableJson.value(Map.of("provider", "local")));
var sandboxes = SandboxBindings.builder()
        .stateStore(records)
        .resolver(binding -> local.target(workspace))
        .canDelete(binding -> allRelatedExecutionsSettled(binding))
        .build();
sandboxes.start();
Executor executor = sandboxes.executor(environment, shellBinding);

bash agentscope-examples/durable/integrated/quickstart/run.sh
bash agentscope-examples/durable/integrated/http-server/run.sh
bash agentscope-examples/durable/integrated/tool-in-sandbox/run.sh
bash agentscope-examples/durable/brain-hands/echo-tool/run.sh --compile-only
bash agentscope-examples/durable/brain-hands/tool-in-sandbox/run.sh --compile-only

var children = new DelegationExecutor(backend, request ->
        new DelegationTarget(
                new ThreadSpec(
                        new ThreadKey(request.session(), "research"),
                        new ThreadBinding("research", "v1", AgentScopeCheckpointCodec.STATE_VERSION)),
                DurableJson.value("research request")));