adam/open-swe
1
0
Fork 0
mirror of https://github.com/Sea-Haven-Industries/open-swe.git synced 2026-10-04 10:12:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: parallelize thread polling (#143)

Browse source 
* fix: parallelize thread polling

* format
This commit is contained in:
Brace Sproul 2025-06-12 13:29:20 -07:00 • committed by GitHub
parent 76d412275a
commit f549d119ab
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 18 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
apps/web/src/lib/polling/thread-poller.ts
View file

@ -55,22 +55,27 @@ export class ThreadPoller {
const changedThreadIds: string[] = []; const changedThreadIds: string[] = [];
const errors: string[] = []; const errors: string[] = [];
for (const currentThread of threadsToPool) { const pollUpdatePromise = Promise.allSettled(
try { threadsToPool.map(async (currentThread) => {
const updatedThread = await this.getThreadFn(currentThread.thread_id); try {
if (updatedThread) { const updatedThread = await this.getThreadFn(
updatedThreads.push(updatedThread); currentThread.thread_id,
);
if (updatedThread) {
updatedThreads.push(updatedThread);
if (this.hasThreadChanged(currentThread, updatedThread)) { if (this.hasThreadChanged(currentThread, updatedThread)) {
changedThreadIds.push(updatedThread.thread_id); changedThreadIds.push(updatedThread.thread_id);
}
} }
} catch (error) {
errors.push(`Thread ${currentThread.thread_id}: ${error}`);
updatedThreads.push(currentThread);
} }
} catch (error) { }),
errors.push(`Thread ${currentThread.thread_id}: ${error}`); );
updatedThreads.push(currentThread); await pollUpdatePromise;
}
}
if (changedThreadIds.length > 0) { if (changedThreadIds.length > 0) {
this.config.onUpdate(updatedThreads, changedThreadIds); this.config.onUpdate(updatedThreads, changedThreadIds);

62
apps/web/src/providers/Thread.tsx
View file

@ -9,10 +9,9 @@ import {
Dispatch, Dispatch,
SetStateAction, SetStateAction,
useEffect, useEffect,
useTransition,
} from "react"; } from "react";
import { createClient } from "./client"; import { createClient } from "./client";
import { TaskPlan, GraphState } from "@open-swe/shared/open-swe/types"; import { GraphState } from "@open-swe/shared/open-swe/types";
import { useThreadPolling } from "@/hooks/useThreadPolling"; import { useThreadPolling } from "@/hooks/useThreadPolling";
interface ThreadContextType { interface ThreadContextType {
@ -22,7 +21,6 @@ interface ThreadContextType {
setThreadsLoading: Dispatch<SetStateAction<boolean>>; setThreadsLoading: Dispatch<SetStateAction<boolean>>;
refreshThreads: () => Promise<void>; refreshThreads: () => Promise<void>;
getThread: (threadId: string) => Promise<Thread<GraphState> | null>; getThread: (threadId: string) => Promise<Thread<GraphState> | null>;
isPending: boolean;
recentlyUpdatedThreads: Set<string>; recentlyUpdatedThreads: Set<string>;
handleThreadClick: ( handleThreadClick: (
thread: Thread<GraphState>, thread: Thread<GraphState>,
@ -43,62 +41,6 @@ function getThreadSearchMetadata(
} }
} }
const getTaskCounts = (
tasks?: TaskPlan,
proposedPlan?: string[],
existingCounts?: { totalTasksCount: number; completedTasksCount: number },
): { totalTasksCount: number; completedTasksCount: number } => {
const defaultCounts = existingCounts || {
totalTasksCount: 0,
completedTasksCount: 0,
};
if (proposedPlan && proposedPlan.length > 0 && !tasks) {
return {
totalTasksCount: proposedPlan.length,
completedTasksCount: 0,
};
}
if (!tasks || !tasks.tasks || tasks.tasks.length === 0) {
return defaultCounts;
}
const activeTaskIndex = tasks.activeTaskIndex;
const activeTask = tasks.tasks.find(
(task) => task.taskIndex === activeTaskIndex,
);
if (
!activeTask ||
!activeTask.planRevisions ||
activeTask.planRevisions.length === 0
) {
return defaultCounts;
}
const activeRevisionIndex = activeTask.activeRevisionIndex;
const activeRevision = activeTask.planRevisions.find(
(revision) => revision.revisionIndex === activeRevisionIndex,
);
if (
!activeRevision ||
!activeRevision.plans ||
activeRevision.plans.length === 0
) {
return defaultCounts;
}
const plans = activeRevision.plans;
const completedTasksCount = plans.filter((p) => p.completed)?.length || 0;
return {
totalTasksCount: plans.length,
completedTasksCount,
};
};
export function ThreadProvider({ children }: { children: ReactNode }) { export function ThreadProvider({ children }: { children: ReactNode }) {
const apiUrl: string | undefined = process.env.NEXT_PUBLIC_API_URL ?? ""; const apiUrl: string | undefined = process.env.NEXT_PUBLIC_API_URL ?? "";
const assistantId: string | undefined = const assistantId: string | undefined =
@ -106,7 +48,6 @@ export function ThreadProvider({ children }: { children: ReactNode }) {
const [threads, setThreads] = useState<Thread<GraphState>[]>([]); const [threads, setThreads] = useState<Thread<GraphState>[]>([]);
const [threadsLoading, setThreadsLoading] = useState(false); const [threadsLoading, setThreadsLoading] = useState(false);
const [isPending, startTransition] = useTransition();
const [recentlyUpdatedThreads, setRecentlyUpdatedThreads] = useState< const [recentlyUpdatedThreads, setRecentlyUpdatedThreads] = useState<
Set<string> Set<string>
>(new Set()); >(new Set());
@ -215,7 +156,6 @@ export function ThreadProvider({ children }: { children: ReactNode }) {
setThreadsLoading, setThreadsLoading,
refreshThreads, refreshThreads,
getThread, getThread,
isPending,
recentlyUpdatedThreads, recentlyUpdatedThreads,
handleThreadClick, handleThreadClick,
}; };