diff --git a/backend-java/src/main/java/io/agentflow/api/controller/GlobalExceptionHandler.java b/backend-java/src/main/java/io/agentflow/api/controller/GlobalExceptionHandler.java index 20f9baa..af2d7dd 100644 --- a/backend-java/src/main/java/io/agentflow/api/controller/GlobalExceptionHandler.java +++ b/backend-java/src/main/java/io/agentflow/api/controller/GlobalExceptionHandler.java @@ -7,6 +7,7 @@ import io.agentflow.api.service.RunComparisonValidationException; import io.agentflow.api.service.RunConflictException; import io.agentflow.api.service.RunNotFoundException; +import io.agentflow.api.service.ThreadNotFoundException; import io.agentflow.api.security.ForbiddenException; import io.agentflow.api.security.UnauthorizedException; import java.util.Map; @@ -33,6 +34,11 @@ public ResponseEntity> handleAgentVersionNotFound( return ResponseEntity.status(HttpStatus.NOT_FOUND).body(Map.of("detail", ex.getMessage())); } + @ExceptionHandler(ThreadNotFoundException.class) + public ResponseEntity> handleThreadNotFound(ThreadNotFoundException ex) { + return ResponseEntity.status(HttpStatus.NOT_FOUND).body(Map.of("detail", ex.getMessage())); + } + @ExceptionHandler(RunNotFoundException.class) public ResponseEntity> handleRunNotFound(RunNotFoundException ex) { return ResponseEntity.status(HttpStatus.NOT_FOUND).body(Map.of("detail", ex.getMessage())); diff --git a/backend-java/src/main/java/io/agentflow/api/controller/ThreadsController.java b/backend-java/src/main/java/io/agentflow/api/controller/ThreadsController.java new file mode 100644 index 0000000..6d94cb0 --- /dev/null +++ b/backend-java/src/main/java/io/agentflow/api/controller/ThreadsController.java @@ -0,0 +1,59 @@ +package io.agentflow.api.controller; + +import io.agentflow.api.dto.RunResponse; +import io.agentflow.api.dto.ThreadCreateRequest; +import io.agentflow.api.dto.ThreadMessageResponse; +import io.agentflow.api.dto.ThreadResponse; +import io.agentflow.api.service.ThreadService; +import jakarta.validation.Valid; +import java.util.List; +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.ResponseStatus; +import org.springframework.web.bind.annotation.RestController; + +@RestController +@RequestMapping("/v1/threads") +public class ThreadsController { + + private final ThreadService service; + + public ThreadsController(ThreadService service) { + this.service = service; + } + + @PostMapping + @ResponseStatus(HttpStatus.CREATED) + public ThreadResponse create(@Valid @RequestBody ThreadCreateRequest payload) { + return service.create(payload); + } + + @GetMapping + public List list(@RequestParam(defaultValue = "50") int limit) { + return service.list(limit); + } + + @GetMapping("/{id}") + public ThreadResponse get(@PathVariable String id) { + return service.get(id); + } + + @GetMapping("/{id}/messages") + public ThreadMessageResponse.Page listMessages( + @PathVariable String id, + @RequestParam(required = false) String cursor, + @RequestParam(defaultValue = "50") int limit) { + return service.listMessages(id, cursor, limit); + } + + @GetMapping("/{id}/runs") + public List listRuns( + @PathVariable String id, @RequestParam(defaultValue = "50") int limit) { + return service.listRuns(id, limit); + } +} diff --git a/backend-java/src/main/java/io/agentflow/api/dto/RunCreateRequest.java b/backend-java/src/main/java/io/agentflow/api/dto/RunCreateRequest.java index f0f6137..c60025d 100644 --- a/backend-java/src/main/java/io/agentflow/api/dto/RunCreateRequest.java +++ b/backend-java/src/main/java/io/agentflow/api/dto/RunCreateRequest.java @@ -19,6 +19,9 @@ public class RunCreateRequest { */ private String adapter; + @JsonProperty("thread_id") + private String threadId; + public String getAgentId() { return agentId; } @@ -50,4 +53,12 @@ public String getAdapter() { public void setAdapter(String adapter) { this.adapter = adapter; } + + public String getThreadId() { + return threadId; + } + + public void setThreadId(String threadId) { + this.threadId = threadId; + } } diff --git a/backend-java/src/main/java/io/agentflow/api/dto/RunResponse.java b/backend-java/src/main/java/io/agentflow/api/dto/RunResponse.java index 7d06858..5137a60 100644 --- a/backend-java/src/main/java/io/agentflow/api/dto/RunResponse.java +++ b/backend-java/src/main/java/io/agentflow/api/dto/RunResponse.java @@ -11,6 +11,7 @@ public class RunResponse { private String id; private String tenantId; private String agentId; + private String threadId; private String adapter; private RunStatus status; private Map input; @@ -74,6 +75,7 @@ public static RunResponse fromEntity( dto.id = entity.getId(); dto.tenantId = entity.getTenantId(); dto.agentId = entity.getAgentId(); + dto.threadId = entity.getThreadId(); dto.adapter = entity.getAdapter(); dto.status = entity.getStatus(); dto.input = entity.getInput(); @@ -101,6 +103,10 @@ public String getAgentId() { return agentId; } + public String getThreadId() { + return threadId; + } + public String getAdapter() { return adapter; } diff --git a/backend-java/src/main/java/io/agentflow/api/dto/ThreadCreateRequest.java b/backend-java/src/main/java/io/agentflow/api/dto/ThreadCreateRequest.java new file mode 100644 index 0000000..2143222 --- /dev/null +++ b/backend-java/src/main/java/io/agentflow/api/dto/ThreadCreateRequest.java @@ -0,0 +1,51 @@ +package io.agentflow.api.dto; + +import com.fasterxml.jackson.annotation.JsonProperty; +import jakarta.validation.constraints.NotBlank; + +public class ThreadCreateRequest { + + @NotBlank + @JsonProperty("agent_id") + private String agentId; + + private String title; + + @JsonProperty("user_id") + private String userId; + + @JsonProperty("project_id") + private String projectId; + + public String getAgentId() { + return agentId; + } + + public void setAgentId(String agentId) { + this.agentId = agentId; + } + + public String getTitle() { + return title; + } + + public void setTitle(String title) { + this.title = title; + } + + public String getUserId() { + return userId; + } + + public void setUserId(String userId) { + this.userId = userId; + } + + public String getProjectId() { + return projectId; + } + + public void setProjectId(String projectId) { + this.projectId = projectId; + } +} diff --git a/backend-java/src/main/java/io/agentflow/api/dto/ThreadMessageResponse.java b/backend-java/src/main/java/io/agentflow/api/dto/ThreadMessageResponse.java new file mode 100644 index 0000000..abaf852 --- /dev/null +++ b/backend-java/src/main/java/io/agentflow/api/dto/ThreadMessageResponse.java @@ -0,0 +1,99 @@ +package io.agentflow.api.dto; + +import io.agentflow.api.entity.MessageEntity; +import java.time.Instant; +import java.util.List; +import java.util.Map; + +public class ThreadMessageResponse { + + private String id; + private String runId; + private int index; + private String stepId; + private String role; + private String name; + private String content; + private String toolCallId; + private Map extra; + private Instant createdAt; + + public static ThreadMessageResponse from(MessageEntity entity, String runId) { + ThreadMessageResponse dto = new ThreadMessageResponse(); + dto.id = entity.getId(); + dto.runId = runId; + dto.index = entity.getIndex(); + dto.stepId = entity.getStepId(); + dto.role = entity.getRole(); + dto.name = entity.getName(); + dto.content = entity.getContent(); + dto.toolCallId = entity.getToolCallId(); + dto.extra = entity.getExtra(); + dto.createdAt = entity.getCreatedAt(); + return dto; + } + + public String getId() { + return id; + } + + public String getRunId() { + return runId; + } + + public int getIndex() { + return index; + } + + public String getStepId() { + return stepId; + } + + public String getRole() { + return role; + } + + public String getName() { + return name; + } + + public String getContent() { + return content; + } + + public String getToolCallId() { + return toolCallId; + } + + public Map getExtra() { + return extra; + } + + public Instant getCreatedAt() { + return createdAt; + } + + public static class Page { + private final List items; + private final String nextCursor; + private final boolean hasMore; + + public Page(List items, String nextCursor, boolean hasMore) { + this.items = items; + this.nextCursor = nextCursor; + this.hasMore = hasMore; + } + + public List getItems() { + return items; + } + + public String getNextCursor() { + return nextCursor; + } + + public boolean isHasMore() { + return hasMore; + } + } +} diff --git a/backend-java/src/main/java/io/agentflow/api/dto/ThreadResponse.java b/backend-java/src/main/java/io/agentflow/api/dto/ThreadResponse.java new file mode 100644 index 0000000..e993e46 --- /dev/null +++ b/backend-java/src/main/java/io/agentflow/api/dto/ThreadResponse.java @@ -0,0 +1,61 @@ +package io.agentflow.api.dto; + +import io.agentflow.api.entity.ThreadEntity; +import java.time.Instant; + +public class ThreadResponse { + + private String id; + private String tenantId; + private String projectId; + private String agentId; + private String userId; + private String title; + private Instant createdAt; + private Instant updatedAt; + + public static ThreadResponse fromEntity(ThreadEntity entity) { + ThreadResponse dto = new ThreadResponse(); + dto.id = entity.getId(); + dto.tenantId = entity.getTenantId(); + dto.projectId = entity.getProjectId(); + dto.agentId = entity.getAgentId(); + dto.userId = entity.getUserId(); + dto.title = entity.getTitle(); + dto.createdAt = entity.getCreatedAt(); + dto.updatedAt = entity.getUpdatedAt(); + return dto; + } + + public String getId() { + return id; + } + + public String getTenantId() { + return tenantId; + } + + public String getProjectId() { + return projectId; + } + + public String getAgentId() { + return agentId; + } + + public String getUserId() { + return userId; + } + + public String getTitle() { + return title; + } + + public Instant getCreatedAt() { + return createdAt; + } + + public Instant getUpdatedAt() { + return updatedAt; + } +} diff --git a/backend-java/src/main/java/io/agentflow/api/entity/RunEntity.java b/backend-java/src/main/java/io/agentflow/api/entity/RunEntity.java index 6d09edf..67857f1 100644 --- a/backend-java/src/main/java/io/agentflow/api/entity/RunEntity.java +++ b/backend-java/src/main/java/io/agentflow/api/entity/RunEntity.java @@ -28,6 +28,9 @@ public class RunEntity { @Column(name = "agent_id", nullable = false, length = 26) private String agentId; + @Column(name = "thread_id", length = 26) + private String threadId; + @Column(nullable = false, length = 64) private String adapter; @@ -101,6 +104,14 @@ public void setAgentId(String agentId) { this.agentId = agentId; } + public String getThreadId() { + return threadId; + } + + public void setThreadId(String threadId) { + this.threadId = threadId; + } + public String getAdapter() { return adapter; } diff --git a/backend-java/src/main/java/io/agentflow/api/entity/ThreadEntity.java b/backend-java/src/main/java/io/agentflow/api/entity/ThreadEntity.java new file mode 100644 index 0000000..0921140 --- /dev/null +++ b/backend-java/src/main/java/io/agentflow/api/entity/ThreadEntity.java @@ -0,0 +1,121 @@ +package io.agentflow.api.entity; + +import com.github.f4b6a3.ulid.UlidCreator; +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Id; +import jakarta.persistence.PrePersist; +import jakarta.persistence.PreUpdate; +import jakarta.persistence.Table; +import java.time.Instant; + +@Entity +@Table(name = "threads") +public class ThreadEntity { + + @Id + @Column(length = 26) + private String id; + + @Column(name = "tenant_id", nullable = false, length = 64) + private String tenantId = "default"; + + @Column(name = "project_id", length = 26) + private String projectId; + + @Column(name = "agent_id", nullable = false, length = 26) + private String agentId; + + @Column(name = "user_id", length = 128) + private String userId; + + @Column(length = 256) + private String title; + + @Column(name = "created_at", nullable = false) + private Instant createdAt; + + @Column(name = "updated_at", nullable = false) + private Instant updatedAt; + + @PrePersist + void onCreate() { + if (id == null) { + id = UlidCreator.getUlid().toString(); + } + Instant now = Instant.now(); + if (createdAt == null) { + createdAt = now; + } + updatedAt = now; + } + + @PreUpdate + void onUpdate() { + updatedAt = Instant.now(); + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getTenantId() { + return tenantId; + } + + public void setTenantId(String tenantId) { + this.tenantId = tenantId; + } + + public String getProjectId() { + return projectId; + } + + public void setProjectId(String projectId) { + this.projectId = projectId; + } + + public String getAgentId() { + return agentId; + } + + public void setAgentId(String agentId) { + this.agentId = agentId; + } + + public String getUserId() { + return userId; + } + + public void setUserId(String userId) { + this.userId = userId; + } + + public String getTitle() { + return title; + } + + public void setTitle(String title) { + this.title = title; + } + + public Instant getCreatedAt() { + return createdAt; + } + + public void setCreatedAt(Instant createdAt) { + this.createdAt = createdAt; + } + + public Instant getUpdatedAt() { + return updatedAt; + } + + public void setUpdatedAt(Instant updatedAt) { + this.updatedAt = updatedAt; + } +} diff --git a/backend-java/src/main/java/io/agentflow/api/repository/RunRepository.java b/backend-java/src/main/java/io/agentflow/api/repository/RunRepository.java index 78023bf..60a3d48 100644 --- a/backend-java/src/main/java/io/agentflow/api/repository/RunRepository.java +++ b/backend-java/src/main/java/io/agentflow/api/repository/RunRepository.java @@ -18,4 +18,6 @@ List findRecentByTenantId( Optional findByIdAndTenantId(String id, String tenantId); List findAllByTenantId(String tenantId); + + List findAllByThreadIdOrderByCreatedAtAsc(String threadId); } diff --git a/backend-java/src/main/java/io/agentflow/api/repository/ThreadRepository.java b/backend-java/src/main/java/io/agentflow/api/repository/ThreadRepository.java new file mode 100644 index 0000000..0bab4a9 --- /dev/null +++ b/backend-java/src/main/java/io/agentflow/api/repository/ThreadRepository.java @@ -0,0 +1,19 @@ +package io.agentflow.api.repository; + +import io.agentflow.api.entity.ThreadEntity; +import java.util.List; +import java.util.Optional; +import org.springframework.data.domain.Pageable; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; + +public interface ThreadRepository extends JpaRepository { + + Optional findByIdAndTenantId(String id, String tenantId); + + @Query( + "SELECT t FROM ThreadEntity t WHERE t.tenantId = :tenantId ORDER BY t.createdAt DESC") + List findRecentByTenantId( + @Param("tenantId") String tenantId, Pageable pageable); +} diff --git a/backend-java/src/main/java/io/agentflow/api/service/RunService.java b/backend-java/src/main/java/io/agentflow/api/service/RunService.java index 7a6fc04..58f628b 100644 --- a/backend-java/src/main/java/io/agentflow/api/service/RunService.java +++ b/backend-java/src/main/java/io/agentflow/api/service/RunService.java @@ -15,6 +15,7 @@ import io.agentflow.api.entity.RunEntity; import io.agentflow.api.entity.RunStatus; import io.agentflow.api.entity.StepEntity; +import io.agentflow.api.entity.ThreadEntity; import io.agentflow.api.entity.ToolCallEntity; import io.agentflow.api.jobs.CancelSignal; import io.agentflow.api.jobs.JobProducer; @@ -22,6 +23,7 @@ import io.agentflow.api.repository.MessageRepository; import io.agentflow.api.repository.RunRepository; import io.agentflow.api.repository.StepRepository; +import io.agentflow.api.repository.ThreadRepository; import io.agentflow.api.repository.ToolCallRepository; import io.agentflow.api.security.AccessControl; import io.agentflow.api.security.Role; @@ -55,6 +57,7 @@ public class RunService { private final ToolCallRepository toolCalls; private final CheckpointRepository checkpoints; private final AgentService agentService; + private final ThreadRepository threads; private final JobProducer jobProducer; private final CancelSignal cancelSignal; @@ -65,6 +68,7 @@ public RunService( ToolCallRepository toolCalls, CheckpointRepository checkpoints, AgentService agentService, + ThreadRepository threads, JobProducer jobProducer, CancelSignal cancelSignal) { this.runs = runs; @@ -73,6 +77,7 @@ public RunService( this.toolCalls = toolCalls; this.checkpoints = checkpoints; this.agentService = agentService; + this.threads = threads; this.jobProducer = jobProducer; this.cancelSignal = cancelSignal; } @@ -85,9 +90,22 @@ public RunResponse create(RunCreateRequest req) { ? req.getAdapter() : agent.getAdapter(); + String threadId = req.getThreadId(); + if (threadId != null && !threadId.isBlank()) { + ThreadEntity thread = threads + .findByIdAndTenantId(threadId, agent.getTenantId()) + .orElseThrow(() -> new ThreadNotFoundException(threadId)); + if (!agent.getId().equals(thread.getAgentId())) { + throw new ThreadNotFoundException(threadId); + } + } else { + threadId = null; + } + RunEntity run = new RunEntity(); run.setTenantId(agent.getTenantId()); run.setAgentId(agent.getId()); + run.setThreadId(threadId); run.setAdapter(adapter); run.setStatus(RunStatus.PENDING); run.setInput(new HashMap<>(req.getInput())); diff --git a/backend-java/src/main/java/io/agentflow/api/service/ThreadNotFoundException.java b/backend-java/src/main/java/io/agentflow/api/service/ThreadNotFoundException.java new file mode 100644 index 0000000..5157e70 --- /dev/null +++ b/backend-java/src/main/java/io/agentflow/api/service/ThreadNotFoundException.java @@ -0,0 +1,8 @@ +package io.agentflow.api.service; + +public class ThreadNotFoundException extends RuntimeException { + + public ThreadNotFoundException(String threadId) { + super("Thread not found: " + threadId); + } +} diff --git a/backend-java/src/main/java/io/agentflow/api/service/ThreadService.java b/backend-java/src/main/java/io/agentflow/api/service/ThreadService.java new file mode 100644 index 0000000..2fc6780 --- /dev/null +++ b/backend-java/src/main/java/io/agentflow/api/service/ThreadService.java @@ -0,0 +1,113 @@ +package io.agentflow.api.service; + +import io.agentflow.api.dto.RunResponse; +import io.agentflow.api.dto.ThreadCreateRequest; +import io.agentflow.api.dto.ThreadMessageResponse; +import io.agentflow.api.dto.ThreadResponse; +import io.agentflow.api.entity.AgentEntity; +import io.agentflow.api.entity.MessageEntity; +import io.agentflow.api.entity.RunEntity; +import io.agentflow.api.entity.ThreadEntity; +import io.agentflow.api.repository.MessageRepository; +import io.agentflow.api.repository.RunRepository; +import io.agentflow.api.repository.ThreadRepository; +import io.agentflow.api.security.AccessControl; +import io.agentflow.api.security.Role; +import java.util.ArrayList; +import java.util.List; +import org.springframework.data.domain.PageRequest; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +@Service +public class ThreadService { + + private final ThreadRepository threads; + private final RunRepository runs; + private final MessageRepository messages; + private final AgentService agentService; + + public ThreadService( + ThreadRepository threads, + RunRepository runs, + MessageRepository messages, + AgentService agentService) { + this.threads = threads; + this.runs = runs; + this.messages = messages; + this.agentService = agentService; + } + + @Transactional + public ThreadResponse create(ThreadCreateRequest req) { + AccessControl.require(Role.OPERATOR); + AgentEntity agent = agentService.getEntity(req.getAgentId()); + ThreadEntity thread = new ThreadEntity(); + thread.setTenantId(agent.getTenantId()); + thread.setAgentId(agent.getId()); + thread.setProjectId(req.getProjectId()); + thread.setUserId(req.getUserId()); + thread.setTitle(req.getTitle()); + return ThreadResponse.fromEntity(threads.save(thread)); + } + + @Transactional(readOnly = true) + public List list(int limit) { + String tenantId = AccessControl.tenantId(Role.VIEWER); + int capped = Math.max(1, Math.min(limit, 200)); + return threads.findRecentByTenantId(tenantId, PageRequest.of(0, capped)).stream() + .map(ThreadResponse::fromEntity) + .toList(); + } + + @Transactional(readOnly = true) + public ThreadResponse get(String id) { + return ThreadResponse.fromEntity(requireThread(id)); + } + + @Transactional(readOnly = true) + public ThreadMessageResponse.Page listMessages(String id, String cursor, int limit) { + ThreadEntity thread = requireThread(id); + int capped = Math.max(1, Math.min(limit, 200)); + List threadRuns = runs.findAllByThreadIdOrderByCreatedAtAsc(thread.getId()); + List all = new ArrayList<>(); + for (RunEntity run : threadRuns) { + for (MessageEntity message : messages.findAllByRunIdOrderByIndexAsc(run.getId())) { + all.add(ThreadMessageResponse.from(message, run.getId())); + } + } + if (cursor != null && !cursor.isBlank()) { + all = all.stream().filter(m -> cursorKey(m).compareTo(cursor) < 0).toList(); + } + boolean hasMore = all.size() > capped; + List page = + hasMore ? all.subList(Math.max(0, all.size() - capped), all.size()) : all; + String nextCursor = hasMore && !page.isEmpty() ? cursorKey(page.get(0)) : null; + return new ThreadMessageResponse.Page(List.copyOf(page), nextCursor, hasMore); + } + + @Transactional(readOnly = true) + public List listRuns(String id, int limit) { + ThreadEntity thread = requireThread(id); + int capped = Math.max(1, Math.min(limit, 200)); + return runs.findAllByThreadIdOrderByCreatedAtAsc(thread.getId()).stream() + .limit(capped) + .map(RunResponse::fromEntity) + .toList(); + } + + ThreadEntity requireThread(String id) { + String tenantId = AccessControl.tenantId(Role.VIEWER); + return threads + .findByIdAndTenantId(id, tenantId) + .orElseThrow(() -> new ThreadNotFoundException(id)); + } + + private static String cursorKey(ThreadMessageResponse message) { + return message.getCreatedAt() + + "|" + + String.format("%08d", message.getIndex()) + + "|" + + message.getId(); + } +} diff --git a/backend-java/src/test/java/io/agentflow/api/service/RunServiceTest.java b/backend-java/src/test/java/io/agentflow/api/service/RunServiceTest.java index 8925756..750ec8e 100644 --- a/backend-java/src/test/java/io/agentflow/api/service/RunServiceTest.java +++ b/backend-java/src/test/java/io/agentflow/api/service/RunServiceTest.java @@ -17,6 +17,7 @@ import io.agentflow.api.repository.MessageRepository; import io.agentflow.api.repository.RunRepository; import io.agentflow.api.repository.StepRepository; +import io.agentflow.api.repository.ThreadRepository; import io.agentflow.api.repository.ToolCallRepository; import java.util.List; import java.util.Map; @@ -78,6 +79,7 @@ private static RunService service(RunRepository runs, AgentService agents) { mock(ToolCallRepository.class), mock(CheckpointRepository.class), agents, + mock(ThreadRepository.class), mock(JobProducer.class), mock(CancelSignal.class)); } diff --git a/backend/alembic/env.py b/backend/alembic/env.py index e0ce27a..c6f05cb 100644 --- a/backend/alembic/env.py +++ b/backend/alembic/env.py @@ -20,8 +20,10 @@ Agent, Checkpoint, Message, + Project, Run, Step, + Thread, ToolCall, ) diff --git a/backend/alembic/versions/0006_threads.py b/backend/alembic/versions/0006_threads.py new file mode 100644 index 0000000..62baf97 --- /dev/null +++ b/backend/alembic/versions/0006_threads.py @@ -0,0 +1,60 @@ +"""Add threads table and runs.thread_id for L1 conversation memory. + +Revision ID: 0006_threads +Revises: 0005_project_scoped_rbac +Create Date: 2026-09-03 +""" + +from __future__ import annotations + +import sqlalchemy as sa + +from alembic import op + +revision = "0006_threads" +down_revision = "0005_project_scoped_rbac" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "threads", + sa.Column("id", sa.String(26), primary_key=True), + sa.Column("tenant_id", sa.String(64), nullable=False), + sa.Column("project_id", sa.String(26), nullable=True), + sa.Column("agent_id", sa.String(26), nullable=False), + sa.Column("user_id", sa.String(128), nullable=True), + sa.Column("title", sa.String(256), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["agent_id"], ["agents.id"], name="fk_threads_agent_id", ondelete="CASCADE" + ), + ) + op.create_index("ix_threads_tenant_id", "threads", ["tenant_id"]) + op.create_index("ix_threads_tenant_agent", "threads", ["tenant_id", "agent_id"]) + op.create_index("ix_threads_project_id", "threads", ["project_id"]) + op.create_index("ix_threads_agent_id", "threads", ["agent_id"]) + + op.add_column("runs", sa.Column("thread_id", sa.String(26), nullable=True)) + op.create_foreign_key( + "fk_runs_thread_id", + "runs", + "threads", + ["thread_id"], + ["id"], + ondelete="SET NULL", + ) + op.create_index("ix_runs_thread_id", "runs", ["thread_id"]) + + +def downgrade() -> None: + op.drop_index("ix_runs_thread_id", table_name="runs") + op.drop_constraint("fk_runs_thread_id", "runs", type_="foreignkey") + op.drop_column("runs", "thread_id") + op.drop_index("ix_threads_agent_id", table_name="threads") + op.drop_index("ix_threads_project_id", table_name="threads") + op.drop_index("ix_threads_tenant_agent", table_name="threads") + op.drop_index("ix_threads_tenant_id", table_name="threads") + op.drop_table("threads") diff --git a/backend/app/adapters/base.py b/backend/app/adapters/base.py index 269b00b..bf0a209 100644 --- a/backend/app/adapters/base.py +++ b/backend/app/adapters/base.py @@ -45,6 +45,9 @@ class AdapterContext: resume: RunResumeContext | None = None # Ordered chat history loaded from the messages table on retry / resume. run_messages: list[dict[str, Any]] | None = None + # Prior turns from the same thread (window-trimmed); empty when no thread. + thread_id: str | None = None + thread_messages: list[dict[str, Any]] = field(default_factory=list) step_index_base: int = 0 emit: EmitCallback = field( default=None # type: ignore[assignment] diff --git a/backend/app/adapters/langgraph_adapter.py b/backend/app/adapters/langgraph_adapter.py index 338335a..3d8ea7d 100644 --- a/backend/app/adapters/langgraph_adapter.py +++ b/backend/app/adapters/langgraph_adapter.py @@ -756,8 +756,14 @@ async def handler(state: dict[str, Any]) -> dict[str, Any]: stream_tokens = run_state.config.get("stream_tokens", True) started = time.monotonic() + history = [ + m + for m in (state.get("messages") or []) + if isinstance(m, dict) and m.get("role") != "system" + ] messages = [ {"role": "system", "content": system_prompt}, + *history, {"role": "user", "content": user_input}, ] response = await self._invoke_model( @@ -1098,11 +1104,24 @@ def _seed_agent_messages( ) -> list[dict[str, Any]]: existing = state.get("messages") if isinstance(existing, list) and existing: - return list(existing) - messages: list[dict[str, Any]] = [{"role": "system", "content": system_prompt}] + messages = list(existing) + else: + messages = [] + + if not any( + isinstance(m, dict) and m.get("role") == "system" for m in messages + ): + messages = [{"role": "system", "content": system_prompt}, *messages] + user_input = str(state.get("input", "")) if user_input: - messages.append({"role": "user", "content": user_input}) + last = messages[-1] if messages else None + if not ( + isinstance(last, dict) + and last.get("role") == "user" + and last.get("content") == user_input + ): + messages.append({"role": "user", "content": user_input}) return messages @@ -1174,6 +1193,8 @@ def _initial_graph_state(ctx: AdapterContext) -> dict[str, Any]: "human_input": None, "route": None, } + if ctx.thread_messages: + default["messages"] = list(ctx.thread_messages) if ctx.resume and ctx.resume.checkpoint_state: saved = ctx.resume.checkpoint_state.get("graph_state") if isinstance(saved, dict): diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index 839202e..31bfa13 100644 --- a/backend/app/api/v1/router.py +++ b/backend/app/api/v1/router.py @@ -1,11 +1,12 @@ from fastapi import APIRouter -from app.api.v1 import agents, events, health, projects, retention, runs +from app.api.v1 import agents, events, health, projects, retention, runs, threads api_router = APIRouter(prefix="/v1") api_router.include_router(health.router) api_router.include_router(projects.router) api_router.include_router(agents.router) +api_router.include_router(threads.router) api_router.include_router(runs.router) api_router.include_router(retention.router) api_router.include_router(events.router) diff --git a/backend/app/api/v1/runs.py b/backend/app/api/v1/runs.py index 5d07019..88067a6 100644 --- a/backend/app/api/v1/runs.py +++ b/backend/app/api/v1/runs.py @@ -17,6 +17,7 @@ RunConflict, RunNotFound, RunService, + ThreadNotFound, ) router = APIRouter(prefix="/runs", tags=["runs"]) @@ -65,6 +66,8 @@ async def create_run( ) except AgentNotFound as exc: raise HTTPException(status_code=404, detail=f"Agent not found: {exc}") from exc + except ThreadNotFound as exc: + raise HTTPException(status_code=404, detail=f"Thread not found: {exc}") from exc await service.start_run(run.id) try: diff --git a/backend/app/api/v1/threads.py b/backend/app/api/v1/threads.py new file mode 100644 index 0000000..1294ae6 --- /dev/null +++ b/backend/app/api/v1/threads.py @@ -0,0 +1,116 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.auth import AuthPrincipal, Role, require_role +from app.db.session import get_session +from app.schemas.run import RunRead, run_read_from_orm +from app.schemas.thread import ( + ThreadCreate, + ThreadMessagePage, + ThreadRead, +) +from app.services.thread_service import ( + AgentNotFound, + ThreadNotFound, + ThreadService, +) + +router = APIRouter(prefix="/threads", tags=["threads"]) + + +def get_thread_service( + session: AsyncSession = Depends(get_session), +) -> ThreadService: + return ThreadService(session=session) + + +@router.post("", response_model=ThreadRead, status_code=status.HTTP_201_CREATED) +async def create_thread( + payload: ThreadCreate, + service: ThreadService = Depends(get_thread_service), + principal: AuthPrincipal = Depends(require_role(Role.OPERATOR)), +) -> ThreadRead: + try: + thread = await service.create_thread( + payload, + tenant_id=principal.tenant_id, + project_id=principal.project_id, + agent_id=principal.agent_id, + ) + except AgentNotFound as exc: + raise HTTPException(status_code=404, detail=f"Agent not found: {exc}") from exc + return ThreadRead.model_validate(thread) + + +@router.get("", response_model=list[ThreadRead]) +async def list_threads( + limit: int = 50, + service: ThreadService = Depends(get_thread_service), + principal: AuthPrincipal = Depends(require_role(Role.VIEWER)), +) -> list[ThreadRead]: + threads = await service.list_threads( + tenant_id=principal.tenant_id, + project_id=principal.project_id, + agent_id=principal.agent_id, + limit=limit, + ) + return [ThreadRead.model_validate(t) for t in threads] + + +@router.get("/{thread_id}", response_model=ThreadRead) +async def get_thread( + thread_id: str, + service: ThreadService = Depends(get_thread_service), + principal: AuthPrincipal = Depends(require_role(Role.VIEWER)), +) -> ThreadRead: + try: + thread = await service.get_thread( + thread_id, + tenant_id=principal.tenant_id, + project_id=principal.project_id, + agent_id=principal.agent_id, + ) + except ThreadNotFound as exc: + raise HTTPException(status_code=404, detail=f"Thread not found: {exc}") from exc + return ThreadRead.model_validate(thread) + + +@router.get("/{thread_id}/messages", response_model=ThreadMessagePage) +async def list_thread_messages( + thread_id: str, + cursor: str | None = None, + limit: int = 50, + service: ThreadService = Depends(get_thread_service), + principal: AuthPrincipal = Depends(require_role(Role.VIEWER)), +) -> ThreadMessagePage: + try: + return await service.list_thread_messages( + thread_id, + cursor=cursor, + limit=limit, + tenant_id=principal.tenant_id, + project_id=principal.project_id, + agent_id=principal.agent_id, + ) + except ThreadNotFound as exc: + raise HTTPException(status_code=404, detail=f"Thread not found: {exc}") from exc + + +@router.get("/{thread_id}/runs", response_model=list[RunRead]) +async def list_thread_runs( + thread_id: str, + limit: int = 50, + service: ThreadService = Depends(get_thread_service), + principal: AuthPrincipal = Depends(require_role(Role.VIEWER)), +) -> list[RunRead]: + try: + runs = await service.list_thread_runs( + thread_id, + tenant_id=principal.tenant_id, + project_id=principal.project_id, + agent_id=principal.agent_id, + limit=limit, + ) + except ThreadNotFound as exc: + raise HTTPException(status_code=404, detail=f"Thread not found: {exc}") from exc + return [run_read_from_orm(run) for run in runs] diff --git a/backend/app/core/config.py b/backend/app/core/config.py index a7c397b..1520ffa 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -272,6 +272,15 @@ class Settings(BaseSettings): le=500, description="Upper bound for limit on GET /v1/runs/{id}/messages.", ) + thread_messages_max: int = Field( + default=40, + ge=0, + le=500, + description=( + "Maximum prior thread messages seeded into AdapterContext " + "(0 = no count cap; token window from agent memory config still applies)." + ), + ) memory_checkpoint_bytes_alert_threshold: int = Field( default=262_144, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index ac2d65b..8b44ae3 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -1,6 +1,7 @@ from app.models.agent import Agent, AgentVersion from app.models.project import Project from app.models.run import Checkpoint, Message, Run, RunStatus, Step, ToolCall +from app.models.thread import Thread __all__ = [ "Agent", @@ -11,5 +12,6 @@ "Run", "RunStatus", "Step", + "Thread", "ToolCall", ] diff --git a/backend/app/models/run.py b/backend/app/models/run.py index 2126741..aa2fc4e 100644 --- a/backend/app/models/run.py +++ b/backend/app/models/run.py @@ -55,6 +55,9 @@ class Run(Base): agent_id: Mapped[str] = mapped_column( ForeignKey("agents.id", ondelete="CASCADE"), index=True ) + thread_id: Mapped[str | None] = mapped_column( + ForeignKey("threads.id", ondelete="SET NULL"), nullable=True, index=True + ) adapter: Mapped[str] = mapped_column(String(64)) status: Mapped[RunStatus] = mapped_column( String(32), default=RunStatus.PENDING, index=True @@ -67,6 +70,7 @@ class Run(Base): metadata_: Mapped[dict[str, Any]] = mapped_column("metadata", JSON, default=dict) agent: Mapped[Agent] = relationship(back_populates="runs") # noqa: F821 + thread: Mapped["Thread | None"] = relationship(back_populates="runs") steps: Mapped[list[Step]] = relationship( back_populates="run", order_by="Step.index", diff --git a/backend/app/models/thread.py b/backend/app/models/thread.py new file mode 100644 index 0000000..6d68ed6 --- /dev/null +++ b/backend/app/models/thread.py @@ -0,0 +1,46 @@ +"""Conversation threads that group Runs for short-term (L1) memory.""" + +from __future__ import annotations + +from sqlalchemy import ForeignKey, Index, String +from sqlalchemy.orm import Mapped, mapped_column, relationship +from ulid import ULID + +from app.db.base import Base +from app.models.agent import DEFAULT_TENANT_ID + + +def _ulid() -> str: + return str(ULID()) + + +class Thread(Base): + """A multi-run conversation scoped to a tenant / agent. + + Runs optionally reference a thread via ``Run.thread_id``. Cross-run + transcripts are assembled by joining messages of all runs in the thread. + """ + + __tablename__ = "threads" + __table_args__ = ( + Index("ix_threads_tenant_id", "tenant_id"), + Index("ix_threads_tenant_agent", "tenant_id", "agent_id"), + Index("ix_threads_project_id", "project_id"), + ) + + id: Mapped[str] = mapped_column(String(26), primary_key=True, default=_ulid) + tenant_id: Mapped[str] = mapped_column( + String(64), default=DEFAULT_TENANT_ID, nullable=False + ) + project_id: Mapped[str | None] = mapped_column(String(26), nullable=True) + agent_id: Mapped[str] = mapped_column( + ForeignKey("agents.id", ondelete="CASCADE"), index=True + ) + user_id: Mapped[str | None] = mapped_column(String(128), nullable=True) + title: Mapped[str | None] = mapped_column(String(256), nullable=True) + + agent: Mapped["Agent"] = relationship() # noqa: F821 + runs: Mapped[list["Run"]] = relationship( # noqa: F821 + back_populates="thread", + order_by="Run.created_at", + ) diff --git a/backend/app/schemas/run.py b/backend/app/schemas/run.py index 5a32a5a..5cbf8ae 100644 --- a/backend/app/schemas/run.py +++ b/backend/app/schemas/run.py @@ -15,6 +15,10 @@ class RunCreate(BaseModel): default=None, description="Override the agent's default adapter for this run.", ) + thread_id: str | None = Field( + default=None, + description="Optional conversation thread; prior turns are seeded into the adapter.", + ) class RunRetry(BaseModel): @@ -102,6 +106,7 @@ class RunRead(BaseModel): tenant_id: str project_id: str | None agent_id: str + thread_id: str | None = None adapter: str status: RunStatus input: dict[str, Any] diff --git a/backend/app/schemas/thread.py b/backend/app/schemas/thread.py new file mode 100644 index 0000000..db47a4a --- /dev/null +++ b/backend/app/schemas/thread.py @@ -0,0 +1,45 @@ +"""Thread (L1 short memory) request/response schemas.""" + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field + +from app.schemas.run import MessageRead + + +class ThreadCreate(BaseModel): + agent_id: str + title: str | None = Field(default=None, max_length=256) + user_id: str | None = Field(default=None, max_length=128) + project_id: str | None = Field( + default=None, + description="Optional; defaults to the agent's project_id when omitted.", + ) + + +class ThreadRead(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: str + tenant_id: str + project_id: str | None + agent_id: str + user_id: str | None + title: str | None + created_at: datetime + updated_at: datetime + + +class ThreadMessageRead(MessageRead): + """A message belonging to a run inside a thread.""" + + run_id: str + + +class ThreadMessagePage(BaseModel): + items: list[ThreadMessageRead] + next_cursor: str | None = Field( + default=None, + description="Opaque cursor for older messages (created_at|index|id).", + ) + has_more: bool = False diff --git a/backend/app/services/run_service.py b/backend/app/services/run_service.py index 713e419..396dd02 100644 --- a/backend/app/services/run_service.py +++ b/backend/app/services/run_service.py @@ -28,6 +28,7 @@ Run, RunStatus, Step, + Thread, ToolCall, ) from app.core.telemetry import ( @@ -67,6 +68,10 @@ class AgentNotFound(Exception): pass +class ThreadNotFound(Exception): + pass + + class RunConflict(Exception): """Run status does not allow the requested control action.""" @@ -120,10 +125,26 @@ async def create_run( ): raise AgentNotFound(payload.agent_id) + thread_id: str | None = payload.thread_id + if thread_id is not None: + thread = await self.session.get(Thread, thread_id) + if ( + thread is None + or thread.tenant_id != agent.tenant_id + or thread.agent_id != agent.id + or ( + project_id is not None + and thread.project_id is not None + and thread.project_id != project_id + ) + ): + raise ThreadNotFound(thread_id) + run = Run( tenant_id=agent.tenant_id, project_id=agent.project_id, agent_id=agent.id, + thread_id=thread_id, adapter=payload.adapter or agent.adapter, status=RunStatus.PENDING, input=payload.input, diff --git a/backend/app/services/thread_service.py b/backend/app/services/thread_service.py new file mode 100644 index 0000000..86169ac --- /dev/null +++ b/backend/app/services/thread_service.py @@ -0,0 +1,242 @@ +"""Thread service — L1 conversation short memory.""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from app.core.config import get_settings +from app.models import Agent, Message, Run, Thread +from app.runtime.memory_window import fit_messages_to_window, parse_memory_config +from app.runtime.messages import message_row_to_dict +from app.schemas.thread import ( + ThreadCreate, + ThreadMessagePage, + ThreadMessageRead, +) + + +class ThreadNotFound(Exception): + def __init__(self, thread_id: str) -> None: + self.thread_id = thread_id + super().__init__(thread_id) + + +class AgentNotFound(Exception): + def __init__(self, agent_id: str) -> None: + self.agent_id = agent_id + super().__init__(agent_id) + + +class ThreadService: + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def create_thread( + self, + payload: ThreadCreate, + *, + tenant_id: str, + project_id: str | None = None, + agent_id: str | None = None, + ) -> Thread: + agent = await self.session.get(Agent, payload.agent_id) + if ( + agent is None + or agent.tenant_id != tenant_id + or (project_id is not None and agent.project_id != project_id) + or (agent_id is not None and agent.id != agent_id) + ): + raise AgentNotFound(payload.agent_id) + + resolved_project = payload.project_id or agent.project_id + if project_id is not None and resolved_project != project_id: + raise AgentNotFound(payload.agent_id) + + thread = Thread( + tenant_id=agent.tenant_id, + project_id=resolved_project, + agent_id=agent.id, + user_id=payload.user_id, + title=payload.title, + ) + self.session.add(thread) + await self.session.commit() + await self.session.refresh(thread) + return thread + + async def get_thread( + self, + thread_id: str, + *, + tenant_id: str | None = None, + project_id: str | None = None, + agent_id: str | None = None, + ) -> Thread: + thread = await self.session.get(Thread, thread_id) + if thread is None: + raise ThreadNotFound(thread_id) + if tenant_id is not None and thread.tenant_id != tenant_id: + raise ThreadNotFound(thread_id) + if project_id is not None and thread.project_id != project_id: + raise ThreadNotFound(thread_id) + if agent_id is not None and thread.agent_id != agent_id: + raise ThreadNotFound(thread_id) + return thread + + async def list_threads( + self, + *, + tenant_id: str, + project_id: str | None = None, + agent_id: str | None = None, + limit: int = 50, + ) -> list[Thread]: + capped = max(1, min(limit, 200)) + stmt = select(Thread).where(Thread.tenant_id == tenant_id) + if project_id is not None: + stmt = stmt.where(Thread.project_id == project_id) + if agent_id is not None: + stmt = stmt.where(Thread.agent_id == agent_id) + stmt = stmt.order_by(Thread.created_at.desc()).limit(capped) + result = await self.session.execute(stmt) + return list(result.scalars().all()) + + async def list_thread_messages( + self, + thread_id: str, + *, + cursor: str | None = None, + limit: int = 50, + tenant_id: str | None = None, + project_id: str | None = None, + agent_id: str | None = None, + ) -> ThreadMessagePage: + await self.get_thread( + thread_id, + tenant_id=tenant_id, + project_id=project_id, + agent_id=agent_id, + ) + capped = max(1, min(limit, get_settings().run_messages_page_max)) + + stmt = ( + select(Message, Run) + .join(Run, Message.run_id == Run.id) + .where(Run.thread_id == thread_id) + .order_by(Run.created_at.asc(), Message.index.asc(), Message.id.asc()) + ) + result = await self.session.execute(stmt) + rows = list(result.all()) + + items = [ + ThreadMessageRead( + id=msg.id, + index=msg.index, + step_id=msg.step_id, + role=msg.role, + name=msg.name, + content=msg.content, + tool_call_id=msg.tool_call_id, + extra=msg.extra or {}, + created_at=msg.created_at, + run_id=run.id, + ) + for msg, run in rows + ] + + # Cursor = exclusive lower bound encoded as created_at|index|id of the + # oldest item on the previous (newer) page. For simplicity we page from + # the end (newest first window) like run messages. + if cursor: + items = [m for m in items if _message_cursor_key(m) < cursor] + + # Newest page: take last N, then report has_more for older ones. + has_more = len(items) > capped + if has_more: + page = items[-capped:] + next_cursor = _message_cursor_key(page[0]) + else: + page = items + next_cursor = None + + return ThreadMessagePage( + items=page, next_cursor=next_cursor, has_more=has_more + ) + + async def list_thread_runs( + self, + thread_id: str, + *, + tenant_id: str | None = None, + project_id: str | None = None, + agent_id: str | None = None, + limit: int = 50, + ) -> list[Run]: + await self.get_thread( + thread_id, + tenant_id=tenant_id, + project_id=project_id, + agent_id=agent_id, + ) + capped = max(1, min(limit, 200)) + stmt = ( + select(Run) + .where(Run.thread_id == thread_id) + .order_by(Run.created_at.asc()) + .limit(capped) + .options(selectinload(Run.steps), selectinload(Run.checkpoints)) + ) + result = await self.session.execute(stmt) + return list(result.scalars().all()) + + async def load_thread_window( + self, + thread_id: str, + *, + exclude_run_id: str | None = None, + agent_config: dict[str, Any] | None = None, + ) -> list[dict[str, Any]]: + """Return OpenAI-style chat dicts for prior thread turns, window-trimmed. + + Skips prompt_echo / system rows so adapters can attach their own system + prompt. Used by the worker when constructing ``AdapterContext``. + """ + stmt = ( + select(Message, Run) + .join(Run, Message.run_id == Run.id) + .where(Run.thread_id == thread_id) + .order_by(Run.created_at.asc(), Run.id.asc(), Message.index.asc(), Message.id.asc()) + ) + if exclude_run_id is not None: + stmt = stmt.where(Run.id != exclude_run_id) + + result = await self.session.execute(stmt) + messages: list[dict[str, Any]] = [] + for msg, _run in result.all(): + extra = msg.extra or {} + if extra.get("kind") == "prompt_echo": + continue + if msg.role == "system": + continue + messages.append(message_row_to_dict(msg)) + + memory_cfg = parse_memory_config(agent_config or {}) + max_messages = get_settings().thread_messages_max + if max_messages > 0 and len(messages) > max_messages: + messages = messages[-max_messages:] + + if memory_cfg.window_tokens > 0: + messages = fit_messages_to_window( + messages, + window_tokens=memory_cfg.window_tokens, + summarize=memory_cfg.summarize, + ) + return messages + + +def _message_cursor_key(msg: ThreadMessageRead) -> str: + return f"{msg.created_at.isoformat()}|{msg.index:08d}|{msg.id}" diff --git a/backend/app/worker/executor.py b/backend/app/worker/executor.py index 522d628..c22b4fd 100644 --- a/backend/app/worker/executor.py +++ b/backend/app/worker/executor.py @@ -135,6 +135,17 @@ async def execute(self, run_id: str, adapter_name: str) -> None: if resume_ctx is not None: run_messages = await service.load_run_messages(run.id) + thread_messages: list[dict[str, Any]] = [] + if run.thread_id: + from app.services.thread_service import ThreadService + + thread_service = ThreadService(session) + thread_messages = await thread_service.load_thread_window( + run.thread_id, + exclude_run_id=run.id, + agent_config=agent_config, + ) + run.status = RunStatus.RUNNING await session.commit() await service._broadcast("run.started", run.id, {}) @@ -155,6 +166,8 @@ async def _emit(event_type: EventType, data: dict[str, Any]) -> None: metadata=run.metadata_, resume=resume_ctx, run_messages=run_messages, + thread_id=run.thread_id, + thread_messages=thread_messages, step_index_base=step_index_base, emit=_emit, ) diff --git a/backend/tests/test_threads.py b/backend/tests/test_threads.py new file mode 100644 index 0000000..5c8115f --- /dev/null +++ b/backend/tests/test_threads.py @@ -0,0 +1,197 @@ +"""M1 Thread short-memory acceptance tests.""" + +from __future__ import annotations + +import asyncio + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.core.config import get_settings + + +async def _poll_until(client, run_id: str, statuses: set[str], *, attempts: int = 80): + body: dict = {} + for _ in range(attempts): + detail = await client.get(f"/v1/runs/{run_id}") + body = detail.json() + if body["status"] in statuses: + return body + await asyncio.sleep(0.05) + return body + + +@pytest.mark.asyncio +async def test_thread_cross_run_messages_and_seed(client, monkeypatch): + """Two runs share a thread; the second AdapterContext gets prior turns.""" + captured: list[list[dict]] = [] + + from app.adapters.base import AdapterContext, AdapterResult, OrchestratorAdapter + from app.adapters import register_adapter + from app.models.run import RunStatus + from ulid import ULID + + adapter_name = f"thread-capture-{ULID()}" + + class CaptureAdapter(OrchestratorAdapter): + name = adapter_name + + async def run(self, ctx: AdapterContext) -> AdapterResult: + captured.append(list(ctx.thread_messages)) + await ctx.emit_step_started(index=0, node="reply") + prompt = str(ctx.input.get("prompt") or "") + await ctx.emit_message(role="user", content=prompt, step_index=0) + reply = f"echo: {prompt}" + await ctx.emit_message(role="assistant", content=reply, step_index=0) + await ctx.emit_step_completed( + index=0, node="reply", output={"reply": reply} + ) + return AdapterResult(status=RunStatus.SUCCEEDED, output={"reply": reply}) + + register_adapter(adapter_name, CaptureAdapter()) + + agent = await client.post( + "/v1/agents", + json={"name": "thread-bot", "adapter": adapter_name, "config": {}}, + ) + assert agent.status_code == 201, agent.text + agent_id = agent.json()["id"] + + thread = await client.post( + "/v1/threads", + json={"agent_id": agent_id, "title": "chat"}, + ) + assert thread.status_code == 201, thread.text + thread_id = thread.json()["id"] + + run1 = await client.post( + "/v1/runs", + json={ + "agent_id": agent_id, + "thread_id": thread_id, + "input": {"prompt": "hello"}, + }, + ) + assert run1.status_code == 202, run1.text + body1 = await _poll_until(client, run1.json()["id"], {"succeeded", "failed"}) + assert body1["status"] == "succeeded" + assert body1["thread_id"] == thread_id + assert captured[0] == [] + + run2 = await client.post( + "/v1/runs", + json={ + "agent_id": agent_id, + "thread_id": thread_id, + "input": {"prompt": "follow-up"}, + }, + ) + assert run2.status_code == 202, run2.text + body2 = await _poll_until(client, run2.json()["id"], {"succeeded", "failed"}) + assert body2["status"] == "succeeded" + assert len(captured) == 2 + roles = [(m["role"], m["content"]) for m in captured[1]] + assert ("user", "hello") in roles + assert ("assistant", "echo: hello") in roles + + msgs = await client.get(f"/v1/threads/{thread_id}/messages") + assert msgs.status_code == 200 + contents = [m["content"] for m in msgs.json()["items"]] + assert "hello" in contents + assert "follow-up" in contents + assert all("run_id" in m for m in msgs.json()["items"]) + + # Without thread_id behavior stays unchanged (empty thread_messages). + run3 = await client.post( + "/v1/runs", + json={"agent_id": agent_id, "input": {"prompt": "solo"}}, + ) + assert run3.status_code == 202 + body3 = await _poll_until(client, run3.json()["id"], {"succeeded", "failed"}) + assert body3["status"] == "succeeded" + assert body3.get("thread_id") is None + assert captured[2] == [] + + +@pytest.mark.asyncio +async def test_thread_cross_tenant_404(monkeypatch): + monkeypatch.setenv("AGENTFLOW_AUTH_ENABLED", "true") + monkeypatch.setenv( + "AGENTFLOW_AUTH_API_KEYS", + "admin-a:tenant-a:admin,admin-b:tenant-b:admin", + ) + get_settings.cache_clear() + + from app.main import app + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://testserver") as client: + async with app.router.lifespan_context(app): + agent = await client.post( + "/v1/agents", + headers={"Authorization": "Bearer admin-a"}, + json={"name": "a-bot", "adapter": "echo", "config": {"delay": 0}}, + ) + assert agent.status_code == 201 + agent_id = agent.json()["id"] + + thread = await client.post( + "/v1/threads", + headers={"Authorization": "Bearer admin-a"}, + json={"agent_id": agent_id}, + ) + assert thread.status_code == 201 + thread_id = thread.json()["id"] + + other = await client.get( + f"/v1/threads/{thread_id}", + headers={"Authorization": "Bearer admin-b"}, + ) + assert other.status_code == 404 + + other_msgs = await client.get( + f"/v1/threads/{thread_id}/messages", + headers={"Authorization": "Bearer admin-b"}, + ) + assert other_msgs.status_code == 404 + + # Creating a run with another tenant's thread_id also 404s. + agent_b = await client.post( + "/v1/agents", + headers={"Authorization": "Bearer admin-b"}, + json={"name": "b-bot", "adapter": "echo", "config": {"delay": 0}}, + ) + denied = await client.post( + "/v1/runs", + headers={"Authorization": "Bearer admin-b"}, + json={ + "agent_id": agent_b.json()["id"], + "thread_id": thread_id, + "input": {"prompt": "x"}, + }, + ) + assert denied.status_code == 404 + + get_settings.cache_clear() + import os + + os.environ.pop("AGENTFLOW_AUTH_ENABLED", None) + os.environ.pop("AGENTFLOW_AUTH_API_KEYS", None) + + +@pytest.mark.asyncio +async def test_missing_thread_on_create_run_404(client): + agent = await client.post( + "/v1/agents", + json={"name": "no-thread-bot", "adapter": "echo", "config": {"delay": 0}}, + ) + agent_id = agent.json()["id"] + resp = await client.post( + "/v1/runs", + json={ + "agent_id": agent_id, + "thread_id": "01INVALIDTHREADID00000000", + "input": {"prompt": "x"}, + }, + ) + assert resp.status_code == 404 diff --git a/docs/api-contract.md b/docs/api-contract.md index 27c36ea..7b76f20 100644 --- a/docs/api-contract.md +++ b/docs/api-contract.md @@ -119,6 +119,42 @@ config added·removed·changed). Copy the snapshot's adapter/config/description onto the live agent as a **new** version (does not delete history). No-op (no bump) when already identical. +### `POST /v1/threads` → 201 + +Request: + +```json +{ + "agent_id": "01HZ...", + "title": "Support chat", + "user_id": "user-42", + "project_id": null +} +``` + +Creates a conversation thread scoped to the agent (and tenant). `title` and +`user_id` are optional. `project_id` defaults to the agent's project when +omitted. Returns 404 if the agent is not visible to the caller. + +### `GET /v1/threads?limit=50` → 200 + +Response: `Thread[]` for the caller's tenant/project/agent scope, newest first. + +### `GET /v1/threads/{id}` → 200 + +Response: one `Thread`. 404 if missing or not visible (including cross-tenant). + +### `GET /v1/threads/{id}/messages?cursor=&limit=50` → 200 + +Merged transcript across all Runs in the thread, ordered by run time then +message index. Each item includes `run_id` plus the usual Message fields. +`cursor` is an opaque string for older pages. 404 if the thread is not visible. + +### `GET /v1/threads/{id}/runs?limit=50` → 200 + +Response: `Run[]` belonging to the thread, oldest first. 404 if the thread is +not visible. + ### `POST /v1/runs` → 202 Request: @@ -128,12 +164,18 @@ Request: "agent_id": "01HZ...", "input": { "prompt": "hi" }, "metadata": {}, - "adapter": "echo" + "adapter": "echo", + "thread_id": "01HZ..." } ``` -`metadata` and `adapter` are optional. When `adapter` is omitted the agent's -default adapter is used. The server replaces the reserved `_agentflow` +`metadata`, `adapter`, and `thread_id` are optional. When `adapter` is omitted +the agent's default adapter is used. When `thread_id` is set, the run is linked +to that thread and the worker seeds prior thread turns into +`AdapterContext.thread_messages` (window-trimmed). Returns 404 if the thread is +missing, belongs to another tenant/agent, or is otherwise not visible. + +The server replaces the reserved `_agentflow` metadata namespace and records the agent's current version as `_agentflow.agent_version`; client-supplied values in that namespace are never trusted. diff --git a/docs/data-model.md b/docs/data-model.md index 90cc4de..1df05ee 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -11,6 +11,8 @@ erDiagram Project ||--o{ Agent : contains Agent ||--o{ Run : has Agent ||--o{ AgentVersion : versions + Agent ||--o{ Thread : has + Thread ||--o{ Run : contains Run ||--o{ Step : has Run ||--o{ Message : has Run ||--o{ Checkpoint : has @@ -40,11 +42,20 @@ erDiagram json config string note } + Thread { + string id PK + string tenant_id + string project_id + string agent_id FK + string user_id + string title + } Run { string id PK string tenant_id string project_id string agent_id FK + string thread_id FK string adapter string status json input @@ -132,6 +143,9 @@ stateDiagram-v2 - **`Message.step_id` is optional** so an adapter can attach a message to a specific node tick when it makes sense, while keeping the run-level ordering authoritative. +- **`Thread` groups Runs for L1 short memory.** `Run.thread_id` is optional; + when set, the worker seeds `AdapterContext.thread_messages` from prior runs + in the same thread (window-trimmed). Messages remain Run-scoped rows. - **`ToolCall.id` is the lifecycle association key.** SSE ``tool_call.started`` / ``tool_call.completed`` carry the same ``call_id`` (equal to ``ToolCall.id``) so parallel or same-name tool invocations within @@ -148,6 +162,10 @@ stateDiagram-v2 | `ix_projects_tenant_id` | list projects for an organization | | `ix_runs_tenant_id` | filter runs by tenant | | `ix_runs_project_id` | apply project-scoped access to runs | +| `ix_runs_thread_id` | list / join runs by conversation thread | +| `ix_threads_tenant_id` | list threads for a tenant | +| `ix_threads_tenant_agent` | filter threads by tenant + agent | +| `ix_threads_agent_id` | cascade / lookup threads by agent | | `ix_runs_tenant_created` | tenant-scoped recent runs | | `ix_runs_status` | filter pending / running runs from a worker | | `ix_runs_agent_id` | list runs for an agent | diff --git a/docs/plan.md b/docs/plan.md index d31f657..245cbcc 100644 --- a/docs/plan.md +++ b/docs/plan.md @@ -2,7 +2,7 @@ 架构与数据模型见 [architecture.md](architecture.md) 与 [data-model.md](data-model.md)。 -**最后核对：** 2026-09-02 +**最后核对：** 2026-09-03 ## 优化方向 @@ -76,7 +76,7 @@ **目标：** 在 runtime 之上提供 eval、记忆与路由，而非塞进 adapter。 - [ ] Run 对比与回归套件 -- [ ] Agent 记忆服务（见下方专项；对话线程 + 情景摘要 + 语义事实 + 文档 RAG） +- [ ] Agent 记忆服务（M1 Thread ✅；M2/M3 情景摘要 + 语义事实 + 文档 RAG 待做） - [ ] 模型路由 / fallback 策略 - [ ] 定时与批量 Run - [ ] 细粒度流式：推理块、多模态附件（DB 持久化） @@ -89,11 +89,11 @@ | 已有 | 实际作用 | 缺口 | | --- | --- | --- | -| `Message`（`backend/app/models/run.py`） | 单次 Run 内有序 transcript；`role` 为 system/user/assistant/tool | 无 `thread_id` / `user_id`；无跨 Run 查询；`content` 仅 TEXT；`GET /v1/runs/{id}` 一次拉全量 | +| `Message`（`backend/app/models/run.py`） | 单次 Run 内有序 transcript；`role` 为 system/user/assistant/tool | 无跨 Run 语义检索；`content` 仅 TEXT（Thread L1 已补跨 Run 窗口） | | `Checkpoint.state` | adapter 不透明快照，供 retry/resume/Temporal 恢复 | LangGraph 每次节点把整份 `graph_state`（含不断增长的 `messages`）写入 JSON；与 `Message` 行重复存储 | -| `AdapterContext`（`backend/app/adapters/base.py`） | 只写：`emit_message` / `emit_checkpoint` | 无 `recall` / `search` / `store`；adapter 看不到历史 Run | -| `POST /v1/runs` | 每次调用独立 Run | 无会话/线程；连续对话只能由客户端自己拼 `input` | -| 控制台 Messages 列表 | 按 `r.messages` 全量渲染 | 无折叠、无检索、无「从哪条记忆注入」的审计 | +| `AdapterContext`（`backend/app/adapters/base.py`） | 只写：`emit_message` / `emit_checkpoint`；只读：`thread_messages`（L1） | 无 `recall` / `search` / `store`（L2/L3 待做） | +| `POST /v1/runs` | 可带可选 `thread_id` 绑定会话 | 无语义记忆；连续对话靠 Thread L1 | +| 控制台 Messages / Threads | Run 级 Messages + Thread 连续对话页 | 无记忆注入审计（M2） | LangGraph 默认图（`backend/app/adapters/langgraph_adapter.py`）把完整 `messages` 列表交给模型，没有窗口裁剪、摘要或 token 预算。retry 时 Java/Python 把 `checkpoint_state` 塞进 `Run.metadata._resume` 再入队 Redis——长对话会把大 JSON 打进 metadata 和 job payload。 @@ -135,14 +135,14 @@ L0 属于执行与审计；L1–L3 才是「Agent Memory」。L0 不能删，但 建议拆成三期，每一期都有独立 API 与验收，避免「记忆服务」变成无边界项目。 -#### M1 — Thread 短记忆（跨 Run 对话） +#### M1 — Thread 短记忆（跨 Run 对话） ✅ **目标：** 同一会话连续 `POST /v1/runs` 时，worker 能自动带上最近对话，而不是让客户端把历史塞进 `input`。 -- 数据：`threads`（`id`, `tenant_id`, `project_id`, `agent_id`, 可选 `user_id`, `title`）+ `Run.thread_id`（nullable FK）。 -- API：`POST /v1/threads`；`POST /v1/runs` 接受 `thread_id`；`GET /v1/threads/{id}/messages`（跨 Run 合并，按时间/index）。 -- Runtime：`AdapterContext` 增加只读 `thread_messages: list[Message]`（已按窗口裁剪）。LangGraph / PydanticAI / AutoGen 从这里 seed，不各自查库。 -- 控制台：按 thread 看连续对话；Run 详情显示所属 thread。 +- [x] 数据：`threads`（`id`, `tenant_id`, `project_id`, `agent_id`, 可选 `user_id`, `title`）+ `Run.thread_id`（nullable FK）。 +- [x] API：`POST /v1/threads`；`POST /v1/runs` 接受 `thread_id`；`GET /v1/threads/{id}/messages`（跨 Run 合并，按时间/index）。 +- [x] Runtime：`AdapterContext` 增加只读 `thread_messages: list[Message]`（已按窗口裁剪）。LangGraph / PydanticAI / AutoGen 从这里 seed，不各自查库。 +- [x] 控制台：按 thread 看连续对话；Run 详情显示所属 thread。 - **非目标：** 向量检索、自动抽事实。 **验收：** 两次 Run 共用 `thread_id`，第二次模型 prompt 含第一次的 user/assistant 回合；不同 `tenant_id` 的 thread 404；不传 `thread_id` 行为与今日一致。 @@ -226,7 +226,7 @@ Adapters **不得** import SQLAlchemy 查 `messages` / `memory_items`（与现 1. L0 优化 1–3（checkpoint 瘦身 + resume 不塞大 blob）——不改 API 契约，风险低。 2. L0 优化 4–6（窗口、分页、去重 emit）——开始动 API 与 LangGraph。 -3. M1 Thread —— 第一个用户可感知的「记忆」。 +3. ~~**M1 Thread**~~ — 已完成：`threads` + `Run.thread_id`；`AdapterContext.thread_messages`；控制台 Threads。 4. M2 MemoryItem + pgvector + `memory.*` 工具 + 注入审计。 5. M3 治理与评测 —— 与回归套件一起。 diff --git a/frontend/app/layout.tsx b/frontend/app/layout.tsx index 056c8a3..367c457 100644 --- a/frontend/app/layout.tsx +++ b/frontend/app/layout.tsx @@ -28,6 +28,9 @@ export default function RootLayout({ Runs + + Threads +
- agent {r.agent_id} · created{" "} - {new Date(r.created_at).toLocaleString()} + agent {r.agent_id} + {r.thread_id ? ( + <> + {" "} + · thread{" "} + + {r.thread_id} + + + ) : null}{" "} + · created {new Date(r.created_at).toLocaleString()}
diff --git a/frontend/app/threads/[id]/page.tsx b/frontend/app/threads/[id]/page.tsx new file mode 100644 index 0000000..41f970c --- /dev/null +++ b/frontend/app/threads/[id]/page.tsx @@ -0,0 +1,123 @@ +"use client"; + +import { useQuery } from "@tanstack/react-query"; +import Link from "next/link"; +import { use } from "react"; + +import { StatusBadge } from "@/components/StatusBadge"; +import { api } from "@/lib/api"; + +interface PageProps { + params: Promise<{ id: string }>; +} + +export default function ThreadDetailPage({ params }: PageProps) { + const { id } = use(params); + + const thread = useQuery({ + queryKey: ["thread", id], + queryFn: () => api.getThread(id), + }); + const messages = useQuery({ + queryKey: ["thread", id, "messages"], + queryFn: () => api.listThreadMessages(id, { limit: 100 }), + }); + const runs = useQuery({ + queryKey: ["thread", id, "runs"], + queryFn: () => api.listThreadRuns(id), + }); + + if (thread.isLoading) return

Loading…

; + if (thread.error || !thread.data) { + return ( +

Failed to load thread: {String(thread.error)}

+ ); + } + + const t = thread.data; + + return ( +
+
+
+
{t.id}
+

+ Thread{t.title ? ` · ${t.title}` : ""} +

+
+ agent {t.agent_id} + {t.user_id ? ( + <> + {" "} + · user {t.user_id} + + ) : null}{" "} + · created {new Date(t.created_at).toLocaleString()} +
+
+ + Back + +
+ +
+

Runs in thread

+ {runs.isLoading ? ( +

Loading…

+ ) : runs.data && runs.data.length > 0 ? ( +
    + {runs.data.map((r) => ( +
  • + + {r.id} + + + + {new Date(r.created_at).toLocaleString()} + +
    • + ))} +
+ ) : ( +

No runs yet.

+ )} +
+ +
+

Conversation

+ {messages.isLoading ? ( +

Loading…

+ ) : messages.data && messages.data.items.length > 0 ? ( +
    + {messages.data.items.map((m) => ( +
  • +
    + {m.role} + + {m.run_id} + + #{m.index} +
    +
    {m.content}
    +
    • + ))} +
+ ) : ( +

No messages yet.

+ )} +
+
+ ); +} diff --git a/frontend/app/threads/page.tsx b/frontend/app/threads/page.tsx new file mode 100644 index 0000000..84acde0 --- /dev/null +++ b/frontend/app/threads/page.tsx @@ -0,0 +1,55 @@ +"use client"; + +import { useQuery } from "@tanstack/react-query"; +import Link from "next/link"; + +import { api } from "@/lib/api"; + +export default function ThreadsPage() { + const threads = useQuery({ + queryKey: ["threads"], + queryFn: api.listThreads, + }); + + return ( +
+
+

Threads

+ cross-run conversation memory +
+ + {threads.isLoading ? ( +

Loading…

+ ) : threads.data && threads.data.length > 0 ? ( + + + + + + + + + + + {threads.data.map((t) => ( + + + + + + + ))} + +
ThreadTitleAgentCreated
+ + {t.id} + + {t.title || "—"}{t.agent_id} + {new Date(t.created_at).toLocaleString()} +
+ ) : ( +

No threads yet.

+ )} +
+ ); +} diff --git a/frontend/lib/api.ts b/frontend/lib/api.ts index 9af0334..af9adb9 100644 --- a/frontend/lib/api.ts +++ b/frontend/lib/api.ts @@ -1,4 +1,12 @@ -import type { Agent, AgentVersion, AgentVersionDiff, MessagePage, Run } from "./types"; +import type { + Agent, + AgentVersion, + AgentVersionDiff, + MessagePage, + Run, + Thread, + ThreadMessagePage, +} from "./types"; import { normalizeUsage } from "./usage"; function authHeaders(): Record { @@ -70,6 +78,7 @@ export const api = { createRun: async (body: { agent_id: string; input: Record; + thread_id?: string; }) => normalizeRun( await request("/v1/runs", { @@ -77,6 +86,37 @@ export const api = { body: JSON.stringify(body), }), ), + listThreads: () => request("/v1/threads"), + getThread: (id: string) => request(`/v1/threads/${id}`), + createThread: (body: { + agent_id: string; + title?: string; + user_id?: string; + }) => + request("/v1/threads", { + method: "POST", + body: JSON.stringify(body), + }), + listThreadMessages: async ( + id: string, + params?: { cursor?: string; limit?: number }, + ) => { + const search = new URLSearchParams(); + if (params?.cursor != null) { + search.set("cursor", params.cursor); + } + if (params?.limit != null) { + search.set("limit", String(params.limit)); + } + const query = search.toString(); + return request( + `/v1/threads/${id}/messages${query ? `?${query}` : ""}`, + ); + }, + listThreadRuns: async (id: string) => { + const runs = await request(`/v1/threads/${id}/runs`); + return runs.map(normalizeRun); + }, listAgents: () => request("/v1/agents"), createAgent: (body: { name: string; diff --git a/frontend/lib/types.ts b/frontend/lib/types.ts index 7c1e289..be5a7fe 100644 --- a/frontend/lib/types.ts +++ b/frontend/lib/types.ts @@ -72,6 +72,7 @@ export interface Run { id: string; tenant_id: string; agent_id: string; + thread_id?: string | null; adapter: string; status: RunStatus; input: Record; @@ -86,6 +87,27 @@ export interface Run { usage: RunUsage; } +export interface Thread { + id: string; + tenant_id: string; + project_id: string | null; + agent_id: string; + user_id: string | null; + title: string | null; + created_at: string; + updated_at: string; +} + +export interface ThreadMessage extends Message { + run_id: string; +} + +export interface ThreadMessagePage { + items: ThreadMessage[]; + next_cursor: string | null; + has_more: boolean; +} + export interface Agent { id: string; tenant_id: string; diff --git a/integrations/autogen/src/agentflow_autogen/adapter.py b/integrations/autogen/src/agentflow_autogen/adapter.py index bd3d0d4..d916614 100644 --- a/integrations/autogen/src/agentflow_autogen/adapter.py +++ b/integrations/autogen/src/agentflow_autogen/adapter.py @@ -83,7 +83,8 @@ async def run(self, ctx: AdapterContext) -> AdapterResult: await handler.begin_step(default_node) await handler.emit_user_prompt(prompt) - async for event in runnable.run_stream(task=prompt): + task = build_autogen_task(prompt, ctx.thread_messages) + async for event in runnable.run_stream(task=task): await handler.handle(event, bridged_names=bridged_names) if bridged_failure: @@ -382,6 +383,30 @@ def prompt_from_input(run_input: dict[str, Any]) -> str: return json.dumps(run_input, ensure_ascii=False) +def build_autogen_task( + prompt: str, thread_messages: list[dict[str, Any]] | None +) -> str | list[TextMessage]: + """Seed AutoGen with prior thread turns when present.""" + if not thread_messages: + return prompt + history: list[TextMessage] = [] + for message in thread_messages: + role = message.get("role") + content = str(message.get("content") or "") + if not content: + continue + if role == "user": + history.append(TextMessage(content=content, source=USER_SOURCE)) + elif role == "assistant": + history.append( + TextMessage(content=content, source=str(message.get("name") or "assistant")) + ) + if not history: + return prompt + history.append(TextMessage(content=prompt, source=USER_SOURCE)) + return history + + def _parse_tool_arguments(raw: Any) -> dict[str, Any]: if isinstance(raw, dict): return raw diff --git a/integrations/pydantic-ai/src/agentflow_pydantic_ai/adapter.py b/integrations/pydantic-ai/src/agentflow_pydantic_ai/adapter.py index eef11a8..5a58900 100644 --- a/integrations/pydantic-ai/src/agentflow_pydantic_ai/adapter.py +++ b/integrations/pydantic-ai/src/agentflow_pydantic_ai/adapter.py @@ -51,6 +51,7 @@ async def run(self, ctx: AdapterContext) -> AdapterResult: try: agent = load_agent(ctx.agent_config.get("agent_factory")) + message_history = openai_dicts_to_pydantic_history(ctx.thread_messages) async with AsyncExitStack() as stack: bridge: AgentFlowToolset | None = None if uses_agentflow_tools(ctx.agent_config): @@ -58,12 +59,15 @@ async def run(self, ctx: AdapterContext) -> AdapterResult: if surface.tools: bridge = AgentFlowToolset(surface, ctx, step_index) - async with agent.run_stream_events( - prompt, - run_id=ctx.run_id, - metadata=dict(ctx.metadata) or None, - toolsets=[bridge] if bridge is not None else None, - ) as events: + run_kwargs: dict[str, Any] = { + "run_id": ctx.run_id, + "metadata": dict(ctx.metadata) or None, + "toolsets": [bridge] if bridge is not None else None, + } + if message_history: + run_kwargs["message_history"] = message_history + + async with agent.run_stream_events(prompt, **run_kwargs) as events: async for event in events: if delta := text_delta(event): await ctx.emit_token_delta(step_index=step_index, delta=delta) @@ -219,6 +223,31 @@ def prompt_from_input(run_input: dict[str, Any]) -> str: return json.dumps(run_input, ensure_ascii=False) +def openai_dicts_to_pydantic_history(messages: list[dict[str, Any]]) -> list[Any]: + """Convert OpenAI-style dicts to PydanticAI ModelMessage history when available.""" + if not messages: + return [] + try: + from pydantic_ai.messages import ( + ModelRequest, + ModelResponse, + TextPart, + UserPromptPart, + ) + except ImportError: + return [] + + history: list[Any] = [] + for message in messages: + role = message.get("role") + content = str(message.get("content") or "") + if role == "user" and content: + history.append(ModelRequest(parts=[UserPromptPart(content=content)])) + elif role == "assistant" and content: + history.append(ModelResponse(parts=[TextPart(content=content)])) + return history + + def adapter_output(value: Any) -> dict[str, Any]: normalized = to_jsonable_python(value, fallback=str) if isinstance(normalized, str):