diff --git a/chat.go b/chat.go index eac1278..e9a3472 100644 --- a/chat.go +++ b/chat.go @@ -283,7 +283,7 @@ func HandleChat(w http.ResponseWriter, r *http.Request) { for iteration := range raw.Iterations { debug("iteration %d of %d", iteration+1, raw.Iterations) - response.Send(StartChunk()) + response.WriteChunk(NewChunk(ChunkStart, nil)) if len(request.Tools) > 0 && iteration == raw.Iterations-1 { debug("no more tool calls") @@ -298,7 +298,7 @@ func HandleChat(w http.ResponseWriter, r *http.Request) { tool, message, err := RunCompletion(ctx, response, request) if err != nil { - response.Send(ErrorChunk(err)) + response.WriteChunk(NewChunk(ChunkError, err)) return } @@ -311,27 +311,27 @@ func HandleChat(w http.ResponseWriter, r *http.Request) { debug("got %q tool call", tool.Name) - response.Send(ToolChunk(tool)) + response.WriteChunk(NewChunk(ChunkTool, tool)) switch tool.Name { case "search_web": err = HandleSearchWebTool(ctx, tool) if err != nil { - response.Send(ErrorChunk(err)) + response.WriteChunk(NewChunk(ChunkError, err)) return } case "fetch_contents": err = HandleFetchContentsTool(ctx, tool) if err != nil { - response.Send(ErrorChunk(err)) + response.WriteChunk(NewChunk(ChunkError, err)) return } case "github_repository": err = HandleGitHubRepositoryTool(ctx, tool) if err != nil { - response.Send(ErrorChunk(err)) + response.WriteChunk(NewChunk(ChunkError, err)) return } @@ -344,14 +344,14 @@ func HandleChat(w http.ResponseWriter, r *http.Request) { debug("finished tool call") - response.Send(ToolChunk(tool)) + response.WriteChunk(NewChunk(ChunkTool, tool)) request.Messages = append(request.Messages, tool.AsAssistantToolCall(message), tool.AsToolMessage(), ) - response.Send(EndChunk()) + response.WriteChunk(NewChunk(ChunkEnd, nil)) } } @@ -386,7 +386,7 @@ func RunCompletion(ctx context.Context, response *Stream, request *openrouter.Ch if id == "" { id = chunk.ID - response.Send(IDChunk(id)) + response.WriteChunk(NewChunk(ChunkID, id)) } if len(chunk.Choices) == 0 { @@ -397,7 +397,7 @@ func RunCompletion(ctx context.Context, response *Stream, request *openrouter.Ch delta := choice.Delta if choice.FinishReason == openrouter.FinishReasonContentFilter { - response.Send(ErrorChunk(errors.New("stopped due to content_filter"))) + response.WriteChunk(NewChunk(ChunkError, errors.New("stopped due to content_filter"))) return nil, "", nil } @@ -434,16 +434,16 @@ func RunCompletion(ctx context.Context, response *Stream, request *openrouter.Ch if delta.Content != "" { buf.WriteString(delta.Content) - response.Send(TextChunk(delta.Content)) + response.WriteChunk(NewChunk(ChunkText, delta.Content)) } else if delta.Reasoning != nil { - response.Send(ReasoningChunk(*delta.Reasoning)) + response.WriteChunk(NewChunk(ChunkReasoning, *delta.Reasoning)) } else if len(delta.Images) > 0 { for _, image := range delta.Images { if image.Type != openrouter.StreamImageTypeImageURL { continue } - response.Send(ImageChunk(image.ImageURL.URL)) + response.WriteChunk(NewChunk(ChunkImage, image.ImageURL.URL)) } } } diff --git a/go.mod b/go.mod index c0e8f7d..0bcac07 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/go-chi/chi/v5 v5.2.3 github.com/goccy/go-yaml v1.18.0 github.com/revrost/go-openrouter v0.2.4 + github.com/vmihailenco/msgpack/v5 v5.4.1 golang.org/x/crypto v0.42.0 ) @@ -16,6 +17,7 @@ require ( github.com/mattn/go-colorable v0.1.14 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/rs/zerolog v1.34.0 // indirect + github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect golang.org/x/sys v0.36.0 // indirect golang.org/x/term v0.35.0 // indirect ) diff --git a/go.sum b/go.sum index b3be62f..d305c3c 100644 --- a/go.sum +++ b/go.sum @@ -29,6 +29,10 @@ github.com/rs/zerolog v1.34.0 h1:k43nTLIwcTVQAncfCw4KZ2VY6ukYoZaBPNOE8txlOeY= github.com/rs/zerolog v1.34.0/go.mod h1:bJsvje4Z08ROH4Nhs5iH600c3IkWhwp44iRc54W6wYQ= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8= +github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok= +github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g= +github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds= golang.org/x/crypto v0.42.0 h1:chiH31gIWm57EkTXpwnqf8qeuMUi0yekh6mT2AvFlqI= golang.org/x/crypto v0.42.0/go.mod h1:4+rDnOTJhQCx2q7/j6rAN5XDw8kPjeaXEUR2eL94ix8= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= diff --git a/static/index.html b/static/index.html index 04d4e7e..6d0b9e2 100644 --- a/static/index.html +++ b/static/index.html @@ -125,8 +125,9 @@ - + + diff --git a/static/js/chat.js b/static/js/chat.js index b89a3c5..b751d6e 100644 --- a/static/js/chat.js +++ b/static/js/chat.js @@ -1,4 +1,15 @@ (() => { + const ChunkType = { + 0: "start", + 1: "id", + 2: "reason", + 3: "text", + 4: "image", + 5: "tool", + 6: "error", + 7: "end", + }; + const $version = document.getElementById("version"), $total = document.getElementById("total"), $title = document.getElementById("title"), @@ -975,50 +986,59 @@ throw new Error(err?.error || response.statusText); } - const reader = response.body.getReader(), - decoder = new TextDecoder(); + const reader = response.body.getReader(); - let buffer = ""; + let buffer = new Uint8Array(); while (true) { const { value, done } = await reader.read(); - if (done) break; + if (done) { + break; + } - buffer += decoder.decode(value, { - stream: true, - }); + const read = new Uint8Array(buffer.length + value.length); - while (true) { - const idx = buffer.indexOf("\n\n"); + read.set(buffer); + read.set(value, buffer.length); - if (idx === -1) { - break; - } + buffer = read; - const frame = buffer.slice(0, idx).trim(); - buffer = buffer.slice(idx + 2); + while (buffer.length >= 5) { + const type = ChunkType[buffer[0]], + length = buffer[1] | (buffer[2] << 8) | (buffer[3] << 16) | (buffer[4] << 24); + + if (!type) { + console.warn("bad chunk type", type); + + buffer = buffer.slice(5 + length); - if (!frame) { continue; } - let chunk; + if (buffer.length < 5 + length) { + break; + } - try { - chunk = JSON.parse(frame); + let data; - if (!chunk) { - throw new Error("invalid chunk"); + if (length > 0) { + const packed = buffer.slice(5, 5 + length); + + try { + data = msgpackr.unpack(packed); + } catch (err) { + console.warn("bad chunk data", packed); + console.warn(err); } - } catch (err) { - console.warn("bad frame", frame); - console.warn(err); } - if (chunk) { - callback(chunk); - } + buffer = buffer.slice(5 + length); + + callback({ + type: type, + data: data, + }); } } } catch (err) { @@ -1030,7 +1050,7 @@ callback({ type: "error", - text: err.message, + data: err.message, }); } finally { callback(aborted ? "aborted" : "done"); @@ -1201,36 +1221,36 @@ break; case "id": - generationID = chunk.text; + generationID = chunk.data; break; case "tool": message.setState("tooling"); - message.setTool(chunk.text); + message.setTool(chunk.data); - if (chunk.text.done) { - totalCost += chunk.text.cost || 0; + if (chunk.data?.done) { + totalCost += chunk.data.cost || 0; finish(); } break; case "image": - message.addImage(chunk.text); + message.addImage(chunk.data); break; case "reason": message.setState("reasoning"); - message.addReasoning(chunk.text); + message.addReasoning(chunk.data); break; case "text": message.setState("receiving"); - message.addText(chunk.text); + message.addText(chunk.data); break; case "error": - message.setError(chunk.text); + message.setError(chunk.data); break; } diff --git a/static/lib/msgpackr.min.js b/static/lib/msgpackr.min.js new file mode 100644 index 0000000..9f0cb0d --- /dev/null +++ b/static/lib/msgpackr.min.js @@ -0,0 +1,2 @@ +!function(e,t){"object"==typeof exports&&"undefined"!=typeof module?t(exports):"function"==typeof define&&define.amd?define(["exports"],t):t((e="undefined"!=typeof globalThis?globalThis:e||self).msgpackr={})}(this,(function(e){"use strict";var t,r,n;try{t=new TextDecoder}catch(e){}var i,s,o,a,u,l=0,f={},c=0,g=0,h=[],p={useRecords:!1,mapsAsObjects:!0};class d{}const y=new d;y.name="MessagePack 0xC1";var w=!1,b=2;try{new Function("")}catch(e){b=1/0}class m{constructor(e){e&&(!1===e.useRecords&&void 0===e.mapsAsObjects&&(e.mapsAsObjects=!0),e.sequential&&!1!==e.trusted&&(e.trusted=!0,e.structures||0==e.useRecords||(e.structures=[],e.maxSharedStructures||(e.maxSharedStructures=0))),e.structures?e.structures.sharedLength=e.structures.length:e.getStructures&&((e.structures=[]).uninitialized=!0,e.structures.sharedLength=0),e.int64AsNumber&&(e.int64AsType="number")),Object.assign(this,e)}unpack(e,t){if(r)return q((()=>(Z(),this?this.unpack(e,t):m.prototype.unpack.call(p,e,t))));e.buffer||e.constructor!==ArrayBuffer||(e="undefined"!=typeof Buffer?Buffer.from(e):new Uint8Array(e)),"object"==typeof t?(n=t.end||e.length,l=t.start||0):(l=0,n=t>-1?t:e.length),g=0,s=null,o=null,r=e;try{u=e.dataView||(e.dataView=new DataView(e.buffer,e.byteOffset,e.byteLength))}catch(t){if(r=null,e instanceof Uint8Array)throw t;throw new Error("Source must be a Uint8Array or Buffer but was a "+(e&&"object"==typeof e?e.constructor.name:typeof e))}if(this instanceof m){if(f=this,this.structures)return i=this.structures,S(t);(!i||i.length>0)&&(i=[])}else f=p,(!i||i.length>0)&&(i=[]);return S(t)}unpackMultiple(e,t){let r,n=0;try{w=!0;let i=e.length,s=this?this.unpack(e,i):Q.unpack(e,i);if(!t){for(r=[s];le.slice(0))));for(let t=0,r=e.length;t=32&&(r.highByte=t-32>>5))}e.sharedLength=e.length;for(let r in t||[])if(r>=0){let n=e[r],i=t[r];i&&(n&&((e.restoreStructures||(e.restoreStructures=[]))[r]=n),e[r]=i)}return this.structures=e}decode(e,t){return this.unpack(e,t)}}function S(e){try{if(!f.trusted&&!w){let e=i.sharedLength||0;en)throw new Error("Unexpected end of MessagePack data");if(!w){let t;try{t=JSON.stringify(e,((e,t)=>"bigint"==typeof t?`${t}n`:t)).slice(0,100)}catch(e){t="(JSON view not available "+e+")"}throw new Error("Data read, but end of buffer not reached "+t)}}return e}catch(e){throw i&&i.restoreStructures&&A(),Z(),(e instanceof RangeError||e.message.startsWith("Unexpected end of buffer")||l>n)&&(e.incomplete=!0),e}}function A(){for(let e in i.restoreStructures)i[e]=i.restoreStructures[e];i.restoreStructures=null}function U(){let e=r[l++];if(e<160){if(e<128){if(e<64)return e;{let t=i[63&e]||f.getStructures&&v()[63&e];return t?(t.read||(t.read=I(t,63&e)),t.read()):e}}if(e<144){if(e-=128,f.mapsAsObjects){let t={};for(let r=0;r=l)return s.slice(l-c,(l+=t)-c);if(0==g&&n<140){let e=t<16?V(t):D(t);if(null!=e)return e}return B(t)}{let t;switch(e){case 192:return null;case 193:return o?(t=U(),t>0?o[1].slice(o.position1,o.position1+=t):o[0].slice(o.position0,o.position0-=t)):y;case 194:return!1;case 195:return!0;case 196:if(t=r[l++],void 0===t)throw new Error("Unexpected end of buffer");return F(t);case 197:return t=u.getUint16(l),l+=2,F(t);case 198:return t=u.getUint32(l),l+=4,F(t);case 199:return N(r[l++]);case 200:return t=u.getUint16(l),l+=2,N(t);case 201:return t=u.getUint32(l),l+=4,N(t);case 202:if(t=u.getFloat32(l),f.useFloat32>2){let e=G[(127&r[l])<<1|r[l+1]>>7];return l+=4,(e*t+(t>0?.5:-.5)>>0)/e}return l+=4,t;case 203:return t=u.getFloat64(l),l+=8,t;case 204:return r[l++];case 205:return t=u.getUint16(l),l+=2,t;case 206:return t=u.getUint32(l),l+=4,t;case 207:return"number"===f.int64AsType?(t=4294967296*u.getUint32(l),t+=u.getUint32(l+4)):"string"===f.int64AsType?t=u.getBigUint64(l).toString():"auto"===f.int64AsType?(t=u.getBigUint64(l),t<=BigInt(2)<=BigInt(-2)<=l?s.slice(l-c,(l+=t)-c):O(t);case 218:return t=u.getUint16(l),g>=(l+=2)?s.slice(l-c,(l+=t)-c):_(t);case 219:return t=u.getUint32(l),g>=(l+=4)?s.slice(l-c,(l+=t)-c):x(t);case 220:return t=u.getUint16(l),l+=2,M(t);case 221:return t=u.getUint32(l),l+=4,M(t);case 222:return t=u.getUint16(l),l+=2,j(t);case 223:return t=u.getUint32(l),l+=4,j(t);default:if(e>=224)return e-256;if(void 0===e){let e=new Error("Unexpected end of MessagePack data");throw e.incomplete=!0,e}throw new Error("Unknown MessagePack token "+e)}}}const E=/^[a-zA-Z_$][a-zA-Z\d_$]*$/;function I(e,t){function r(){if(r.count++>b){let r=e.read=new Function("r","return function(){return "+(f.freezeData?"Object.freeze":"")+"({"+e.map((e=>"__proto__"===e?"__proto_:r()":E.test(e)?e+":r()":"["+JSON.stringify(e)+"]:r()")).join(",")+"})}")(U);return 0===e.highByte&&(e.read=k(t,e.read)),r()}let n={};for(let t=0,r=e.length;tfunction(){let n=r[l++];if(0===n)return t();let s=e<32?-(e+(n<<5)):e+(n<<5),o=i[s]||v()[s];if(!o)throw new Error("Record id is not defined for "+s);return o.read||(o.read=I(o,e)),o.read()};function v(){let e=q((()=>(r=null,f.getStructures())));return i=f._mergeStructures(e,i)}var B=T,O=T,_=T,x=T;function T(e){let n;if(e<16&&(n=V(e)))return n;if(e>64&&t)return t.decode(r.subarray(l,l+=e));const i=l+e,s=[];for(n="";l65535&&(t-=65536,s.push(t>>>10&1023|55296),t=56320|1023&t),s.push(t)}else s.push(e);s.length>=4096&&(n+=R.apply(String,s),s.length=0)}return s.length>0&&(n+=R.apply(String,s)),n}function M(e){let t=new Array(e);for(let r=0;r0)return void(l=t);n[i]=e}return R.apply(String,n)}function V(e){if(e<4){if(e<2){if(0===e)return"";{let e=r[l++];return(128&e)>1?void(l-=1):R(e)}}{let t=r[l++],n=r[l++];if((128&t)>0||(128&n)>0)return void(l-=2);if(e<3)return R(t,n);let i=r[l++];return(128&i)>0?void(l-=3):R(t,n,i)}}{let t=r[l++],n=r[l++],i=r[l++],s=r[l++];if((128&t)>0||(128&n)>0||(128&i)>0||(128&s)>0)return void(l-=4);if(e<6){if(4===e)return R(t,n,i,s);{let e=r[l++];return(128&e)>0?void(l-=5):R(t,n,i,s,e)}}if(e<8){let o=r[l++],a=r[l++];if((128&o)>0||(128&a)>0)return void(l-=6);if(e<7)return R(t,n,i,s,o,a);let u=r[l++];return(128&u)>0?void(l-=7):R(t,n,i,s,o,a,u)}{let o=r[l++],a=r[l++],u=r[l++],f=r[l++];if((128&o)>0||(128&a)>0||(128&u)>0||(128&f)>0)return void(l-=8);if(e<10){if(8===e)return R(t,n,i,s,o,a,u,f);{let e=r[l++];return(128&e)>0?void(l-=9):R(t,n,i,s,o,a,u,f,e)}}if(e<12){let c=r[l++],g=r[l++];if((128&c)>0||(128&g)>0)return void(l-=10);if(e<11)return R(t,n,i,s,o,a,u,f,c,g);let h=r[l++];return(128&h)>0?void(l-=11):R(t,n,i,s,o,a,u,f,c,g,h)}{let c=r[l++],g=r[l++],h=r[l++],p=r[l++];if((128&c)>0||(128&g)>0||(128&h)>0||(128&p)>0)return void(l-=12);if(e<14){if(12===e)return R(t,n,i,s,o,a,u,f,c,g,h,p);{let e=r[l++];return(128&e)>0?void(l-=13):R(t,n,i,s,o,a,u,f,c,g,h,p,e)}}{let d=r[l++],y=r[l++];if((128&d)>0||(128&y)>0)return void(l-=14);if(e<15)return R(t,n,i,s,o,a,u,f,c,g,h,p,d,y);let w=r[l++];return(128&w)>0?void(l-=15):R(t,n,i,s,o,a,u,f,c,g,h,p,d,y,w)}}}}}function L(){let e,t=r[l++];if(t<192)e=t-160;else switch(t){case 217:e=r[l++];break;case 218:e=u.getUint16(l),l+=2;break;case 219:e=u.getUint32(l),l+=4;break;default:throw new Error("Expected string")}return T(e)}function F(e){return f.copyBuffers?Uint8Array.prototype.slice.call(r,l,l+=e):r.subarray(l,l+=e)}function N(e){let t=r[l++];if(h[t]){let n;return h[t](r.subarray(l,n=l+=e),(e=>{l=e;try{return U()}finally{l=n}}))}throw new Error("Unknown extension type "+t)}var P=new Array(4096);function z(){let e=r[l++];if(!(e>=160&&e<192))return l--,C(U());if(e-=160,g>=l)return s.slice(l-c,(l+=e)-c);if(!(0==g&&n<180))return B(e);let t,i=4095&(e<<5^(e>1?u.getUint16(l):e>0?r[l]:0)),o=P[i],a=l,f=l+e-3,h=0;if(o&&o.bytes==e){for(;a["string","number","boolean","bigint"].includes(typeof e))))return e.flat().toString();throw new Error("Invalid property type for record: "+typeof e)}const W=(e,t)=>{let r=U().map(C),n=e;void 0!==t&&(e=e<32?-((t<<5)+e):(t<<5)+e,r.highByte=t);let s=i[e];return s&&(s.isShared||w)&&((i.restoreStructures||(i.restoreStructures=[]))[e]=s),i[e]=r,r.read=I(r,n),r.read()};h[0]=()=>{},h[0].noBuffer=!0,h[66]=e=>{let t=e.byteLength%8||8,r=BigInt(128&e[0]?e[0]-256:e[0]);for(let n=1;n{let r=t-e;if(r<=40){let r=n.getBigUint64(e);for(let i=e+8;i>4<<3),o=i(e,s),a=i(s,t);return o<{let e=U();if(!J[e[0]]){let t=Error(e[1],{cause:e[2]});return t.name=e[0],t}return J[e[0]](e[1],{cause:e[2]})},h[105]=e=>{if(!1===f.structuredClone)throw new Error("Structured clone extension is disabled");let t=u.getUint32(l-4);a||(a=new Map);let n,i=r[l];n=i>=144&&i<160||220==i||221==i?[]:i>=128&&i<144||222==i||223==i?new Map:(i>=199&&i<=201||i>=212&&i<=216)&&115===r[l+1]?new Set:{};let s={target:n};a.set(t,s);let o=U();if(!s.used)return s.target=o;if(Object.assign(n,o),n instanceof Map)for(let[e,t]of o.entries())n.set(e,t);if(n instanceof Set)for(let e of Array.from(o))n.add(e);return n},h[112]=e=>{if(!1===f.structuredClone)throw new Error("Structured clone extension is disabled");let t=u.getUint32(l-4),r=a.get(t);return r.used=!0,r.target},h[115]=()=>new Set(U());const $=["Int8","Uint8","Uint8Clamped","Int16","Uint16","Int32","Uint32","Float32","Float64","BigInt64","BigUint64"].map((e=>e+"Array"));let K="object"==typeof globalThis?globalThis:window;h[116]=e=>{let t=e[0],r=Uint8Array.prototype.slice.call(e,1).buffer,n=$[t];if(!n){if(16===t)return r;if(17===t)return new DataView(r);throw new Error("Could not find typed array for code "+t)}return new K[n](r)},h[120]=()=>{let e=U();return new RegExp(e[0],e[1])};const Y=[];function q(e){let t=n,h=l,p=c,d=g,y=s,b=a,m=o,S=new Uint8Array(r.slice(0,n)),A=i,U=i.slice(0,i.length),E=f,I=w,k=e();return n=t,l=h,c=p,g=d,s=y,a=b,o=m,r=S,w=I,(i=A).splice(0,i.length,...U),f=E,u=new DataView(r.buffer,r.byteOffset,r.byteLength),k}function Z(){r=null,a=null,i=null}h[98]=e=>{let t=(e[0]<<24)+(e[1]<<16)+(e[2]<<8)+e[3],r=l;return l+=t-e.length,o=Y,(o=[L(),L()]).position0=0,o.position1=0,o.postBundlePosition=l,l=r,U()},h[255]=e=>4==e.length?new Date(1e3*(16777216*e[0]+(e[1]<<16)+(e[2]<<8)+e[3])):8==e.length?new Date(((e[0]<<22)+(e[1]<<14)+(e[2]<<6)+(e[3]>>2))/1e6+1e3*(4294967296*(3&e[3])+16777216*e[4]+(e[5]<<16)+(e[6]<<8)+e[7])):12==e.length?new Date(((e[0]<<24)+(e[1]<<16)+(e[2]<<8)+e[3])/1e6+1e3*((128&e[4]?-281474976710656:0)+1099511627776*e[6]+4294967296*e[7]+16777216*e[8]+(e[9]<<16)+(e[10]<<8)+e[11])):new Date("invalid");const G=new Array(147);for(let e=0;e<256;e++)G[e]=+("1e"+Math.floor(45.15-.30103*e));const H=m;var Q=new m({useRecords:!1});const X=Q.unpack,ee=Q.unpackMultiple,te=Q.unpack,re={NEVER:0,ALWAYS:1,DECIMAL_ROUND:3,DECIMAL_FIT:4};let ne,ie,se,oe=new Float32Array(1),ae=new Uint8Array(oe.buffer,0,4);try{ne=new TextEncoder}catch(e){}const ue="undefined"!=typeof Buffer,le=ue?function(e){return Buffer.allocUnsafeSlow(e)}:Uint8Array,fe=ue?Buffer:Uint8Array,ce=ue?4294967296:2144337920;let ge,he,pe,de,ye=0,we=null;const be=/[\u0080-\uFFFF]/,me=Symbol("record-id");class Se extends m{constructor(e){let t,r,n,i;super(e),this.offset=0;let s=fe.prototype.utf8Write?function(e,t){return ge.utf8Write(e,t,ge.byteLength-t)}:!(!ne||!ne.encodeInto)&&function(e,t){return ne.encodeInto(e,ge.subarray(t)).written},o=this;e||(e={});let a=e&&e.sequential,u=e.structures||e.saveStructures,l=e.maxSharedStructures;if(null==l&&(l=u?32:0),l>8160)throw new Error("Maximum maxSharedStructure is 8160");e.structuredClone&&null==e.moreTypes&&(this.moreTypes=!0);let f=e.maxOwnStructures;null==f&&(f=u?32:64),this.structures||0==e.useRecords||(this.structures=[]);let c=l>32||f+l>64,g=l+64,h=l+f+64;if(h>8256)throw new Error("Maximum maxSharedStructure + maxOwnStructure is 8192");let p=[],d=0,y=0;this.pack=this.encode=function(e,s){if(ge||(ge=new le(8192),pe=ge.dataView||(ge.dataView=new DataView(ge.buffer,0,8192)),ye=0),de=ge.length-10,de-ye<2048?(ge=new le(ge.length),pe=ge.dataView||(ge.dataView=new DataView(ge.buffer,0,ge.length)),de=ge.length-10,ye=0):ye=ye+7&2147483640,t=ye,s&De&&(ye+=255&s),i=o.structuredClone?new Map:null,o.bundleStrings&&"string"!=typeof e?(we=[],we.size=1/0):we=null,n=o.structures,n){n.uninitialized&&(n=o._mergeStructures(o.getStructures()));let e=n.sharedLength||0;if(e>l)throw new Error("Shared structures is larger than maximum shared structures, try increasing maxSharedStructures to "+n.sharedLength);if(!n.transitions){n.transitions=Object.create(null);for(let t=0;te.offset>t.offset?1:-1)),r=e.length,n=-1;for(;a&&r>0;){let i=e[--r].offset+t;ia.position+t?n>=0&&(n+=6):(n>=0&&(pe.setUint32(a.position+t,pe.getUint32(a.position+t)+n),n=-1),a=a.previous,r++)}n>=0&&a&&pe.setUint32(a.position+t,pe.getUint32(a.position+t)+n),ye+=6*e.length,ye>de&&I(ye),o.offset=ye;let s=function(e,t){let r,n=6*t.length,i=e.length-n;for(;r=t.pop();){let t=r.offset,s=r.id;e.copyWithin(t+n,t,i),n-=6;let o=t+n;e[o++]=214,e[o++]=105,e[o++]=s>>24,e[o++]=s>>16&255,e[o++]=s>>8&255,e[o++]=255&s,i=t}return e}(ge.subarray(t,ye),e);return i=null,s}return o.offset=ye,s&je?(ge.start=t,ge.end=ye,ge):ge.subarray(t,ye)}catch(e){throw u=e,e}finally{if(n&&(w(),r&&o.saveStructures)){let r=n.sharedLength||0,i=ge.subarray(t,ye),a=function(e,t){return e.isCompatible=e=>{let r=!e||(t.lastNamedStructuresLength||0)===e.length;return r||t._mergeStructures(e),r},e}(n,o);if(!u)return!1===o.saveStructures(a,a.isCompatible)?o.pack(e,s):(o.lastNamedStructuresLength=r,ge.length>1073741824&&(ge=null),i)}ge.length>1073741824&&(ge=null),s&Re&&(ye=t)}};const w=()=>{y<10&&y++;let e=n.sharedLength||0;if(n.length>e&&!a&&(n.length=e),d>1e4)n.transitions=null,y=0,d=0,p.length>0&&(p=[]);else if(p.length>0&&!a){for(let e=0,t=p.length;e{var t=e.length;t<16?ge[ye++]=144|t:t<65536?(ge[ye++]=220,ge[ye++]=t>>8,ge[ye++]=255&t):(ge[ye++]=221,pe.setUint32(ye,t),ye+=4);for(let r=0;r{ye>de&&(ge=I(ye));var r,n=typeof e;if("string"===n){let n,i=e.length;if(we&&i>=4&&i<4096){if((we.size+=i)>21760){let e,r,n=(we[0]?3*we[0].length+we[1].length:0)+10;ye+n>de&&(ge=I(ye+n)),we.position?(r=we,ge[ye]=200,ye+=3,ge[ye++]=98,e=ye-t,ye+=4,Ie(t,m,0),pe.setUint16(e+t-3,ye-t-e)):(ge[ye++]=214,ge[ye++]=98,e=ye-t,ye+=4),we=["",""],we.previous=r,we.size=0,we.position=e}let r=be.test(e);return we[r?0:1]+=e,ge[ye++]=193,void m(r?-i:i)}n=i<32?1:i<256?2:i<65536?3:5;let o=3*i;if(ye+o>de&&(ge=I(ye+o)),i<64||!s){let t,s,o,a=ye+n;for(t=0;t>6|192,ge[a++]=63&s|128):55296==(64512&s)&&56320==(64512&(o=e.charCodeAt(t+1)))?(s=65536+((1023&s)<<10)+(1023&o),t++,ge[a++]=s>>18|240,ge[a++]=s>>12&63|128,ge[a++]=s>>6&63|128,ge[a++]=63&s|128):(ge[a++]=s>>12|224,ge[a++]=s>>6&63|128,ge[a++]=63&s|128);r=a-ye-n}else r=s(e,ye+n);r<32?ge[ye++]=160|r:r<256?(n<2&&ge.copyWithin(ye+2,ye+1,ye+1+r),ge[ye++]=217,ge[ye++]=r):r<65536?(n<3&&ge.copyWithin(ye+3,ye+2,ye+2+r),ge[ye++]=218,ge[ye++]=r>>8,ge[ye++]=255&r):(n<5&&ge.copyWithin(ye+5,ye+3,ye+3+r),ge[ye++]=219,pe.setUint32(ye,r),ye+=4),ye+=r}else if("number"===n)if(e>>>0===e)e<32||e<128&&!1===this.useRecords||e<64&&!this.randomAccessStructure?ge[ye++]=e:e<256?(ge[ye++]=204,ge[ye++]=e):e<65536?(ge[ye++]=205,ge[ye++]=e>>8,ge[ye++]=255&e):(ge[ye++]=206,pe.setUint32(ye,e),ye+=4);else if(e>>0===e)e>=-32?ge[ye++]=256+e:e>=-128?(ge[ye++]=208,ge[ye++]=e+256):e>=-32768?(ge[ye++]=209,pe.setInt16(ye,e),ye+=2):(ge[ye++]=210,pe.setInt32(ye,e),ye+=4);else{let t;if((t=this.useFloat32)>0&&e<4294967296&&e>=-2147483648){let r;if(ge[ye++]=202,pe.setFloat32(ye,e),t<4||(r=e*G[(127&ge[ye])<<1|ge[ye+1]>>7])>>0===r)return void(ye+=4);ye--}ge[ye++]=203,pe.setFloat64(ye,e),ye+=8}else if("object"===n||"function"===n)if(e){if(i){let r=i.get(e);if(r){if(!r.id){let e=i.idsToInsert||(i.idsToInsert=[]);r.id=e.push(r)}return ge[ye++]=214,ge[ye++]=112,pe.setUint32(ye,r.id),void(ye+=4)}i.set(e,{offset:ye-t})}let s=e.constructor;if(s===Object)E(e);else if(s===Array)b(e);else if(s===Map)if(this.mapAsEmptyObject)ge[ye++]=128;else{(r=e.size)<16?ge[ye++]=128|r:r<65536?(ge[ye++]=222,ge[ye++]=r>>8,ge[ye++]=255&r):(ge[ye++]=223,pe.setUint32(ye,r),ye+=4);for(let[t,r]of e)m(t),m(r)}else{for(let t=0,r=ie.length;t(ge=i,i=null,ye+=e,ye>de&&I(ye),{target:ge,targetView:pe,position:ye-e})),m)}finally{i&&(ge=i,pe=s,ye=o,de=ge.length-10)}return void(n&&(n.length+ye>de&&I(n.length+ye),ye=Ee(n,ge,ye,r.type)))}}if(Array.isArray(e))b(e);else{if(e.toJSON){const t=e.toJSON();if(t!==e)return m(t)}if("function"===n)return m(this.writeFunction&&this.writeFunction(e));E(e)}}}else ge[ye++]=192;else if("boolean"===n)ge[ye++]=e?195:194;else if("bigint"===n){if(e<0x8000000000000000&&e>=-0x8000000000000000)ge[ye++]=211,pe.setBigInt64(ye,e);else if(e<0x10000000000000000&&e>0)ge[ye++]=207,pe.setBigUint64(ye,e);else{if(!this.largeBigIntToFloat){if(this.largeBigIntToString)return m(e.toString());if(this.useBigIntExtension||this.moreTypes){let t,r=e<0?BigInt(-1):BigInt(0);if(e>>BigInt(65536)===r){let n=BigInt(0x10000000000000000)-BigInt(1),i=[];for(;i.push(e&n),e>>BigInt(63)!==r;)e>>=BigInt(64);t=new Uint8Array(new BigUint64Array(i).buffer),t.reverse()}else{let r=e<0,n=(r?~e:e).toString(16);if(n.length%2?n="0"+n:parseInt(n.charAt(0),16)>=8&&(n="00"+n),ue)t=Buffer.from(n,"hex");else{t=new Uint8Array(n.length/2);for(let e=0;ede&&I(t.length+ye),void(ye=Ee(t,ge,ye,66))}throw new RangeError(e+" was too large to fit in MessagePack 64-bit integer format, use useBigIntExtension, or set largeBigIntToFloat to convert to float-64, or set largeBigIntToString to convert to string")}ge[ye++]=203,pe.setFloat64(ye,Number(e))}ye+=8}else{if("undefined"!==n)throw new Error("Unknown type: "+n);this.encodeUndefinedAsNil?ge[ye++]=192:(ge[ye++]=212,ge[ye++]=0,ge[ye++]=0)}},S=this.variableMapSize||this.coercibleKeyAsNumber||this.skipValues?e=>{let t;if(this.skipValues){t=[];for(let r in e)"function"==typeof e.hasOwnProperty&&!e.hasOwnProperty(r)||this.skipValues.includes(e[r])||t.push(r)}else t=Object.keys(e);let r,n=t.length;if(n<16?ge[ye++]=128|n:n<65536?(ge[ye++]=222,ge[ye++]=n>>8,ge[ye++]=255&n):(ge[ye++]=223,pe.setUint32(ye,n),ye+=4),this.coercibleKeyAsNumber)for(let i=0;i{ge[ye++]=222;let r=ye-t;ye+=2;let n=0;for(let t in e)("function"!=typeof e.hasOwnProperty||e.hasOwnProperty(t))&&(m(t),m(e[t]),n++);if(n>65535)throw new Error('Object is too large to serialize with fast 16-bit map size, use the "variableMapSize" option to serialize this object');ge[r+++t]=n>>8,ge[r+t]=255&n},A=!1===this.useRecords?S:e.progressiveRecords&&!c?e=>{let r,i,s=n.transitions||(n.transitions=Object.create(null)),o=ye++-t;for(let a in e)if("function"!=typeof e.hasOwnProperty||e.hasOwnProperty(a)){if(r=s[a],r)s=r;else{let u=Object.keys(e),l=s;s=n.transitions;let f=0;for(let e=0,t=u.length;e{let t,r=n.transitions||(n.transitions=Object.create(null)),i=0;for(let n in e)("function"!=typeof e.hasOwnProperty||e.hasOwnProperty(n))&&(t=r[n],t||(t=r[n]=Object.create(null),i++),r=t);let s=r[me];s?s>=96&&c?(ge[ye++]=96+(31&(s-=96)),ge[ye++]=s>>5):ge[ye++]=s:k(r,r.__keys__||Object.keys(e),i);for(let t in e)("function"!=typeof e.hasOwnProperty||e.hasOwnProperty(t))&&m(e[t])},U="function"==typeof this.useRecords&&this.useRecords,E=U?e=>{U(e)?A(e):S(e)}:A,I=e=>{let r;if(e>16777216){if(e-t>ce)throw new Error("Packed buffer would be larger than maximum buffer size");r=Math.min(ce,4096*Math.round(Math.max((e-t)*(e>67108864?1.25:2),4194304)/4096))}else r=1+(Math.max(e-t<<2,ge.length-1)>>12)<<12;let n=new le(r);return pe=n.dataView||(n.dataView=new DataView(n.buffer,0,r)),e=Math.min(e,ge.length),ge.copy?ge.copy(n,0,t,e):n.set(ge.slice(t,e)),ye-=t,t=0,de=n.length-10,ge=n},k=(e,t,i)=>{let s=n.nextId;s||(s=64),s=h&&(s=g),n.nextId=s+1);let o=t.highByte=s>=96&&c?s-96>>5:-1;e[me]=s,e.__keys__=t,n[s-64]=t,s=0?(ge[ye++]=96+(31&s),ge[ye++]=o):ge[ye++]=s):(o>=0?(ge[ye++]=213,ge[ye++]=114,ge[ye++]=96+(31&s),ge[ye++]=o):(ge[ye++]=212,ge[ye++]=114,ge[ye++]=s),i&&(d+=y*i),p.length>=f&&(p.shift()[me]=0),p.push(e),m(t))},v=(e,r,n,i)=>{let s=ge,o=ye,a=de,u=t;ge=he,ye=0,t=0,ge||(he=ge=new le(8192)),de=ge.length-10,k(e,r,i),he=ge;let l=ye;if(ge=s,ye=o,de=a,t=u,l>1){let e=ye+l-1;e>de&&I(e);let r=n+t;ge.copyWithin(r+l,r+1,ye),ge.set(he.slice(0,l),r),ye=e}else ge[n+t]=he[0]},B=e=>{let i=undefined(e,ge,t,ye,n,I,((e,t,n)=>{if(n)return r=!0;ye=t;let i=ge;return m(e),w(),i!==ge?{position:ye,targetView:pe,target:ge}:ye}),this);if(0===i)return E(e);ye=i}}useBuffer(e){ge=e,ge.dataView||(ge.dataView=new DataView(ge.buffer,ge.byteOffset,ge.byteLength)),pe=ge.dataView,ye=0}set position(e){ye=e}get position(){return ye}clearSharedData(){this.structures&&(this.structures=[]),this.typedStructs&&(this.typedStructs=[])}}function Ae(e,t,r,n){let i=e.byteLength;if(i+1<256){var{target:s,position:o}=r(4+i);s[o++]=199,s[o++]=i+1}else if(i+1<65536){var{target:s,position:o}=r(5+i);s[o++]=200,s[o++]=i+1>>8,s[o++]=i+1&255}else{var{target:s,position:o,targetView:a}=r(7+i);s[o++]=201,a.setUint32(o,i+1),o+=4}s[o++]=116,s[o++]=t,e.buffer||(e=new Uint8Array(e)),s.set(new Uint8Array(e.buffer,e.byteOffset,e.byteLength),o)}function Ue(e,t){let r=e.byteLength;var n,i;if(r<256){var{target:n,position:i}=t(r+2);n[i++]=196,n[i++]=r}else if(r<65536){var{target:n,position:i}=t(r+3);n[i++]=197,n[i++]=r>>8,n[i++]=255&r}else{var{target:n,position:i,targetView:s}=t(r+5);n[i++]=198,s.setUint32(i,r),i+=4}n.set(e,i)}function Ee(e,t,r,n){let i=e.length;switch(i){case 1:t[r++]=212;break;case 2:t[r++]=213;break;case 4:t[r++]=214;break;case 8:t[r++]=215;break;case 16:t[r++]=216;break;default:i<256?(t[r++]=199,t[r++]=i):i<65536?(t[r++]=200,t[r++]=i>>8,t[r++]=255&i):(t[r++]=201,t[r++]=i>>24,t[r++]=i>>16&255,t[r++]=i>>8&255,t[r++]=255&i)}return t[r++]=n,t.set(e,r),r+=i}function Ie(e,t,r){if(we.length>0){pe.setUint32(we.position+e,ye+r-we.position-e),we.stringsPosition=ye-e;let n=we;we=null,t(n[0]),t(n[1])}}se=[Date,Set,Error,RegExp,ArrayBuffer,Object.getPrototypeOf(Uint8Array.prototype).constructor,DataView,d],ie=[{pack(e,t,r){let n=e.getTime()/1e3;if((this.useTimestamp32||0===e.getMilliseconds())&&n>=0&&n<4294967296){let{target:e,targetView:r,position:i}=t(6);e[i++]=214,e[i++]=255,r.setUint32(i,n)}else if(n>0&&n<4294967296){let{target:r,targetView:i,position:s}=t(10);r[s++]=215,r[s++]=255,i.setUint32(s,4e6*e.getMilliseconds()+(n/1e3/4294967296>>0)),i.setUint32(s+4,n)}else if(isNaN(n)){if(this.onInvalidDate)return t(0),r(this.onInvalidDate());let{target:e,targetView:n,position:i}=t(3);e[i++]=212,e[i++]=255,e[i++]=255}else{let{target:r,targetView:i,position:s}=t(15);r[s++]=199,r[s++]=12,r[s++]=255,i.setUint32(s,1e6*e.getMilliseconds()),i.setBigInt64(s+4,BigInt(Math.floor(n)))}}},{pack(e,t,r){if(this.setAsEmptyObject)return t(0),r({});let n=Array.from(e),{target:i,position:s}=t(this.moreTypes?3:0);this.moreTypes&&(i[s++]=212,i[s++]=115,i[s++]=0),r(n)}},{pack(e,t,r){let{target:n,position:i}=t(this.moreTypes?3:0);this.moreTypes&&(n[i++]=212,n[i++]=101,n[i++]=0),r([e.name,e.message,e.cause])}},{pack(e,t,r){let{target:n,position:i}=t(this.moreTypes?3:0);this.moreTypes&&(n[i++]=212,n[i++]=120,n[i++]=0),r([e.source,e.flags])}},{pack(e,t){this.moreTypes?Ae(e,16,t):Ue(ue?Buffer.from(e):new Uint8Array(e),t)}},{pack(e,t){let r=e.constructor;r!==fe&&this.moreTypes?Ae(e,$.indexOf(r.name),t):Ue(e,t)}},{pack(e,t){this.moreTypes?Ae(e,17,t):Ue(ue?Buffer.from(e):new Uint8Array(e),t)}},{pack(e,t){let{target:r,position:n}=t(1);r[n]=193}}];let ke=new Se({useRecords:!1});const ve=ke.pack,Be=ke.pack,Oe=Se,{NEVER:_e,ALWAYS:xe,DECIMAL_ROUND:Te,DECIMAL_FIT:Me}=re,je=512,Re=1024,De=2048;const Ve=function(e,t={}){if(!e||"object"!=typeof e)throw new Error("first argument must be an Iterable, Async Iterable, Iterator, Async Iterator, or a promise");const r=new m(t);let n;const i=e=>{let t;n&&(e=Buffer.concat([n,e]),n=void 0);try{t=r.unpackMultiple(e)}catch(r){if(!r.incomplete)throw r;n=e.slice(r.lastPosition),t=r.values}return t};return"function"==typeof e[Symbol.iterator]?function*(){for(const t of e)yield*i(t)}():"function"==typeof e[Symbol.asyncIterator]?async function*(){for await(const t of e)yield*i(t)}():void 0},Le=function(e,t={}){if(e&&"object"==typeof e){if("function"==typeof e[Symbol.iterator])return function*(e,t){const r=new Se(t);for(const t of e)yield r.pack(t)}(e,t);if("function"==typeof e.then||"function"==typeof e[Symbol.asyncIterator])return async function*(e,t){const r=new Se(t);for await(const t of e)yield r.pack(t)}(e,t);throw new Error("first argument must be an Iterable, Async Iterable, Iterator, Async Iterator, or a Promise")}throw new Error("first argument must be an Iterable, Async Iterable, or a Promise for an Async Iterable")};e.ALWAYS=xe,e.C1=y,e.DECIMAL_FIT=Me,e.DECIMAL_ROUND=Te,e.Decoder=H,e.Encoder=Oe,e.FLOAT32_OPTIONS=re,e.NEVER=_e,e.Packr=Se,e.RESERVE_START_SPACE=De,e.RESET_BUFFER_MODE=Re,e.REUSE_BUFFER_MODE=je,e.Unpackr=m,e.addExtension=function(e){if(e.Class){if(!e.pack&&!e.write)throw new Error("Extension has no pack or write function");if(e.pack&&!e.type)throw new Error("Extension has no type (numeric code to identify the extension)");se.unshift(e.Class),ie.unshift(e)}!function(e){e.unpack?h[e.type]=e.unpack:h[e.type]=e}(e)},e.clearSource=Z,e.decode=te,e.decodeIter=Ve,e.encode=Be,e.encodeIter=Le,e.isNativeAccelerationEnabled=!1,e.mapsAsObjects=!0,e.pack=ve,e.roundFloat32=function(e){oe[0]=e;let t=G[(127&ae[3])<<1|ae[2]>>7];return(t*e+(e>0?.5:-.5)>>0)/t},e.unpack=X,e.unpackMultiple=ee,e.useRecords=!1})); +//# sourceMappingURL=index.min.js.map diff --git a/stream.go b/stream.go index 1ac62e5..0a944a1 100644 --- a/stream.go +++ b/stream.go @@ -3,17 +3,31 @@ package main import ( "bytes" "context" - "encoding/json" + "encoding/binary" "errors" "net/http" "sync" "github.com/revrost/go-openrouter" + "github.com/vmihailenco/msgpack/v5" ) +const ( + ChunkStart ChunkType = 0 + ChunkID ChunkType = 1 + ChunkReasoning ChunkType = 2 + ChunkText ChunkType = 3 + ChunkImage ChunkType = 4 + ChunkTool ChunkType = 5 + ChunkError ChunkType = 6 + ChunkEnd ChunkType = 7 +) + +type ChunkType uint8 + type Chunk struct { - Type string `json:"type"` - Text any `json:"text,omitempty"` + Type ChunkType + Data any } type Stream struct { @@ -46,63 +60,10 @@ func NewStream(w http.ResponseWriter, ctx context.Context) (*Stream, error) { }, nil } -func (s *Stream) Send(ch Chunk) error { - debugIf(ch.Type == "error", "error: %v", ch.Text) - - return WriteChunk(s.wr, s.ctx, ch) -} - -func StartChunk() Chunk { - return Chunk{ - Type: "start", - } -} - -func IDChunk(id string) Chunk { - return Chunk{ - Type: "id", - Text: id, - } -} - -func ReasoningChunk(text string) Chunk { - return Chunk{ - Type: "reason", - Text: text, - } -} - -func TextChunk(text string) Chunk { - return Chunk{ - Type: "text", - Text: CleanChunk(text), - } -} - -func ImageChunk(image string) Chunk { - return Chunk{ - Type: "image", - Text: image, - } -} - -func ToolChunk(tool *ToolCall) Chunk { - return Chunk{ - Type: "tool", - Text: tool, - } -} - -func ErrorChunk(err error) Chunk { - return Chunk{ - Type: "error", - Text: GetErrorMessage(err), - } -} - -func EndChunk() Chunk { - return Chunk{ - Type: "end", +func NewChunk(typ ChunkType, data any) *Chunk { + return &Chunk{ + Type: typ, + Data: data, } } @@ -114,32 +75,43 @@ func GetErrorMessage(err error) string { return err.Error() } -func WriteChunk(w http.ResponseWriter, ctx context.Context, chunk any) error { - if err := ctx.Err(); err != nil { +func (s *Stream) WriteChunk(chunk *Chunk) error { + debugIf(chunk.Type == ChunkError, "error: %v", chunk.Data) + + if err := s.ctx.Err(); err != nil { return err } buf := GetFreeBuffer() defer pool.Put(buf) - if err := json.NewEncoder(buf).Encode(chunk); err != nil { + binary.Write(buf, binary.LittleEndian, chunk.Type) + + if chunk.Data != nil { + data, err := msgpack.Marshal(chunk.Data) + if err != nil { + return err + } + + binary.Write(buf, binary.LittleEndian, uint32(len(data))) + + buf.Write(data) + } else { + binary.Write(buf, binary.LittleEndian, uint32(0)) + } + + if _, err := s.wr.Write(buf.Bytes()); err != nil { return err } - buf.Write([]byte("\n\n")) - - if _, err := w.Write(buf.Bytes()); err != nil { - return err - } - - flusher, ok := w.(http.Flusher) + flusher, ok := s.wr.(http.Flusher) if !ok { return errors.New("failed to create flusher") } select { - case <-ctx.Done(): - return ctx.Err() + case <-s.ctx.Done(): + return s.ctx.Err() default: flusher.Flush()