toki
ac242e5fac
- Add Attempt, Reason, OccurredAt fields to notification.TaskEvent - Populate workflow context in scheduler event emission via parseTaskEventContext - Extend proto-socket task.status.changed payload with additive fields - Update contracts note with new event payload shape - Archive notification_model subtask to agent-task/archive - Fix Plane compose path in core domain rules
85 lines
2.4 KiB
Go
85 lines
2.4 KiB
Go
package protosocket
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/nomadcode/nomadcode-core/internal/notification"
|
|
)
|
|
|
|
// SuccessResponse builds a response envelope correlated to a request.
|
|
func SuccessResponse(req Envelope, payload map[string]any) Envelope {
|
|
return Envelope{
|
|
ProtocolVersion: ProtocolVersion,
|
|
ID: generateID(),
|
|
CorrelationID: req.ID,
|
|
Type: "response",
|
|
Channel: req.Channel,
|
|
Action: req.Action,
|
|
Payload: payload,
|
|
}
|
|
}
|
|
|
|
// ErrorResponse builds an error envelope correlated to a request.
|
|
func ErrorResponse(req Envelope, code, message string, retryable bool) Envelope {
|
|
return Envelope{
|
|
ProtocolVersion: ProtocolVersion,
|
|
ID: generateID(),
|
|
CorrelationID: req.ID,
|
|
Type: "error",
|
|
Channel: req.Channel,
|
|
Action: req.Action,
|
|
Error: &EnvelopeError{
|
|
Code: code,
|
|
Message: message,
|
|
Retryable: retryable,
|
|
},
|
|
}
|
|
}
|
|
|
|
// Broadcaster publishes an envelope to all connected clients.
|
|
type Broadcaster interface {
|
|
BroadcastEnvelope(ctx context.Context, env Envelope) error
|
|
}
|
|
|
|
// TaskEventBroadcaster adapts notification task events into proto-socket
|
|
// task.status.changed events. It implements notification.TaskEventSink.
|
|
type TaskEventBroadcaster struct {
|
|
broadcaster Broadcaster
|
|
}
|
|
|
|
func NewTaskEventBroadcaster(broadcaster Broadcaster) *TaskEventBroadcaster {
|
|
return &TaskEventBroadcaster{broadcaster: broadcaster}
|
|
}
|
|
|
|
func (b *TaskEventBroadcaster) HandleTaskEvent(ctx context.Context, event notification.TaskEvent) error {
|
|
return b.broadcaster.BroadcastEnvelope(ctx, TaskStatusChangedEnvelope(event))
|
|
}
|
|
|
|
// TaskStatusChangedEnvelope maps a notification task event onto the
|
|
// task.status.changed event envelope.
|
|
func TaskStatusChangedEnvelope(event notification.TaskEvent) Envelope {
|
|
payload := map[string]any{
|
|
"id": event.TaskID,
|
|
"type": string(event.Type),
|
|
"status": event.Status,
|
|
"title": event.Title,
|
|
"message": event.Message,
|
|
"attempt": event.Attempt,
|
|
}
|
|
if event.Reason != "" {
|
|
payload["reason"] = event.Reason
|
|
}
|
|
if !event.OccurredAt.IsZero() {
|
|
payload["occurred_at"] = event.OccurredAt.UTC().Format(time.RFC3339Nano)
|
|
}
|
|
|
|
return Envelope{
|
|
ProtocolVersion: ProtocolVersion,
|
|
ID: generateID(),
|
|
Type: "event",
|
|
Channel: "task",
|
|
Action: "task.status.changed",
|
|
Payload: payload,
|
|
}
|
|
}
|