mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
zeekayandHanzo Dev
679ae61163
The engine took a PORT and the SDK an Options.HostPort, so neither end
could use a unix socket for traffic that never leaves the host. The
field name was not the only obstacle: luxfi/zap bound net.Listen("tcp",
":port") and reached a peer with net.DialTimeout("tcp", addr), so a path
was refused at the dialer with "missing port in address" and could not
be bound at all. That is fixed upstream (luxfi/zap Address + Network);
this pins the commit until it is tagged.
EmbedConfig.Address, Embedded.Address, ServeGated(addr) and
client.Options.Address each take ONE address, because a port is a detail
of one transport rather than a thing every listener has. A path binds or
dials a unix socket; anything else is host:port.
TestDialRequiresHostPort became TestDialRequiresAddress: Dial needs an
address, and asserts a socket path is refused for being absent rather
than for being portless. TestE2E_UnixSocket runs a real workflow over a
socket end to end, and TestE2E_UnixSocketRebind proves the socket is
unlinked on stop so a restart binds cleanly.
Co-authored-by: Hanzo Dev <[email protected]>
104 lines
2.9 KiB
Go
104 lines
2.9 KiB
Go
// Copyright © 2026 Hanzo AI. MIT License.
|
|
|
|
// Native realtime event tail for Go callers. Streams Events from
|
|
// http://<host>:<httpPort>/v1/tasks/events without spinning up an
|
|
// HTTP browser stack — just a long-lived TCP request parsing the
|
|
// SSE framing inline.
|
|
//
|
|
// The transport is HTTP/SSE today because that's the canonical wire
|
|
// for the same stream the browser consumes. A pure-ZAP streaming
|
|
// opcode (0x00B1, reserved) lands when luxfi/zap exposes a streaming
|
|
// frame primitive on its Node API.
|
|
//
|
|
// No gRPC. No protobuf. No upstream client.
|
|
|
|
package client
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
)
|
|
|
|
// Event mirrors pkg/tasks.Event on the wire.
|
|
type Event struct {
|
|
Kind string `json:"kind"`
|
|
Namespace string `json:"namespace,omitempty"`
|
|
WorkflowID string `json:"workflow_id,omitempty"`
|
|
RunID string `json:"run_id,omitempty"`
|
|
ScheduleID string `json:"schedule_id,omitempty"`
|
|
BatchID string `json:"batch_id,omitempty"`
|
|
At string `json:"at"`
|
|
Data any `json:"data,omitempty"`
|
|
}
|
|
|
|
// SubscribeEvents opens an SSE stream to the daemon's /v1/tasks/events
|
|
// endpoint and returns a channel that yields decoded Events until ctx
|
|
// is canceled or the connection drops. Cancel ctx to stop.
|
|
//
|
|
// httpAddr is the daemon's HTTP listener (e.g. "127.0.0.1:7243").
|
|
// Pass empty to use http://<Address host>:7243 — the canonical sibling
|
|
// port shipped by tasksd alongside ZAP :9999.
|
|
func SubscribeEvents(ctx context.Context, httpAddr string) (<-chan Event, error) {
|
|
if httpAddr == "" {
|
|
return nil, errors.New("hanzo/tasks/client.SubscribeEvents: httpAddr required")
|
|
}
|
|
if !strings.HasPrefix(httpAddr, "http") {
|
|
httpAddr = "http://" + httpAddr
|
|
}
|
|
url := strings.TrimRight(httpAddr, "/") + "/v1/tasks/events"
|
|
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
req.Header.Set("Accept", "text/event-stream")
|
|
req.Header.Set("Cache-Control", "no-cache")
|
|
|
|
resp, err := http.DefaultClient.Do(req)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if resp.StatusCode != http.StatusOK {
|
|
_ = resp.Body.Close()
|
|
return nil, fmt.Errorf("hanzo/tasks/client.SubscribeEvents: %s → %d", url, resp.StatusCode)
|
|
}
|
|
|
|
out := make(chan Event, 16)
|
|
go func() {
|
|
defer close(out)
|
|
defer resp.Body.Close()
|
|
scanner := bufio.NewScanner(resp.Body)
|
|
// SSE frames are well under any sensible cap; bump the buffer
|
|
// to 256 KiB to handle large workflow data payloads.
|
|
scanner.Buffer(make([]byte, 0, 64*1024), 256*1024)
|
|
for scanner.Scan() {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
line := scanner.Text()
|
|
if !strings.HasPrefix(line, "data: ") {
|
|
continue
|
|
}
|
|
payload := strings.TrimPrefix(line, "data: ")
|
|
if payload == "" {
|
|
continue
|
|
}
|
|
var ev Event
|
|
if err := json.Unmarshal([]byte(payload), &ev); err != nil {
|
|
continue
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case out <- ev:
|
|
}
|
|
}
|
|
}()
|
|
return out, nil
|
|
}
|