sune-org/ORP
1
0
Fork 0
mirror of https://github.com/sune-org/ORP.git synced 2026-01-13 16:17:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feat: Support streaming and replaying image generation

Browse Source
This commit is contained in:
multipleof4
2025-11-16 10:09:37 -08:00
parent f61c8664f7
commit fa17061e9c
1 changed files with 13 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
index.js
View File

@@ -142,7 +142,7 @@ export class MyDurableObject {
} }
replay(ws, after) { replay(ws, after) {
this.buffer.forEach(it => { if (it.seq > after) this.send(ws, { type: 'delta', seq: it.seq, text: it.text }); }); this.buffer.forEach(it => { if (it.seq > after) this.send(ws, { type: 'delta', ...it }); });
if (this.phase === 'done') this.send(ws, { type: 'done' }); if (this.phase === 'done') this.send(ws, { type: 'done' });
else if (['error', 'evicted'].includes(this.phase)) this.send(ws, { type: 'err', message: this.error || 'The run was terminated unexpectedly.' }); else if (['error', 'evicted'].includes(this.phase)) this.send(ws, { type: 'err', message: this.error || 'The run was terminated unexpectedly.' });
} }
@@ -150,8 +150,9 @@ export class MyDurableObject {
flush(force = false) { flush(force = false) {
if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; } if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; }
if (this.pending) { if (this.pending) {
this.buffer.push({ seq: ++this.seq, text: this.pending }); const item = { seq: ++this.seq, text: this.pending };
this.bcast({ type: 'delta', seq: this.seq, text: this.pending }); this.buffer.push(item);
this.bcast({ type: 'delta', ...item });
this.pending = ''; this.pending = '';
this.lastFlushedAt = Date.now(); this.lastFlushedAt = Date.now();
} }
@@ -179,10 +180,11 @@ export class MyDurableObject {
if (req.method === 'GET') { if (req.method === 'GET') {
await this.autopsy(); await this.autopsy();
const text = this.buffer.map(it => it.text).join('') + this.pending; const text = this.buffer.filter(it => it.text).map(it => it.text).join('') + this.pending;
const images = this.buffer.flatMap(it => it.images || []);
const isTerminal = ['done', 'error', 'evicted'].includes(this.phase); const isTerminal = ['done', 'error', 'evicted'].includes(this.phase);
const isError = ['error', 'evicted'].includes(this.phase); const isError = ['error', 'evicted'].includes(this.phase);
const payload = { rid: this.rid, seq: this.seq, phase: this.phase, done: isTerminal, error: isError ? (this.error || 'The run was terminated unexpectedly.') : null, text }; const payload = { rid: this.rid, seq: this.seq, phase: this.phase, done: isTerminal, error: isError ? (this.error || 'The run was terminated unexpectedly.') : null, text, images };
return this.corsJSON(payload); return this.corsJSON(payload);
} }
return this.corsJSON({ error: 'not allowed' }, 405); return this.corsJSON({ error: 'not allowed' }, 405);
@@ -341,6 +343,12 @@ export class MyDurableObject {
this.queueDelta(delta.reasoning); this.queueDelta(delta.reasoning);
hasReasoning = true; hasReasoning = true;
} }
if (Array.isArray(delta?.images) && delta.images.length > 0) {
this.flush(false);
const item = { seq: ++this.seq, images: delta.images };
this.buffer.push(item);
this.bcast({ type: 'delta', ...item });
}
if (delta?.content) { if (delta?.content) {
if (hasReasoning && !hasContent) this.queueDelta('\n'); if (hasReasoning && !hasContent) this.queueDelta('\n');
this.queueDelta(delta.content); this.queueDelta(delta.content);