| 1 | diff --git a/internal/daemon/client_browse.go b/internal/daemon/client_browse.go |
| 2 | index d67a87a..fb13995 100644 |
| 3 | --- a/internal/daemon/client_browse.go |
| 4 | +++ b/internal/daemon/client_browse.go |
| 5 | @@ -34,7 +34,7 @@ type clientBrowseRequest struct { |
| 6 | } |
| 7 | |
| 8 | func clientBrowseMethod(method string) bool { |
| 9 | - return wikiBrowseMethod(method) || approvalBrowseMethod(method) || containsString([]string{"browserepos", "browserepo", "browsefiles", "browsefile", "browsestatus", "browseissues", "browsepulls", "browseissue", "browsepull", "browseagent"}, method) |
| 10 | + return wikiBrowseMethod(method) || approvalBrowseMethod(method) || jobBrowseMethod(method) || containsString([]string{"browserepos", "browserepo", "browsefiles", "browsefile", "browsestatus", "browseissues", "browsepulls", "browseissue", "browsepull", "browseagent"}, method) |
| 11 | } |
| 12 | |
| 13 | func (t *Tenant) browseRead(ctx context.Context, actor string) error { |
| 14 | @@ -64,6 +64,9 @@ func (t *Tenant) executeBrowse(ctx context.Context, actor, method string, params |
| 15 | if approvalBrowseMethod(method) { |
| 16 | return t.executeApprovals(ctx, actor, method, params) |
| 17 | } |
| 18 | + if jobBrowseMethod(method) { |
| 19 | + return t.executeJobs(ctx, actor, method, params) |
| 20 | + } |
| 21 | q := clientBrowseRequest{} |
| 22 | if len(params) > 0 { |
| 23 | if err := json.Unmarshal(params[0], &q); err != nil { |
| 24 | diff --git a/internal/daemon/jobs_browse.go b/internal/daemon/jobs_browse.go |
| 25 | new file mode 100644 |
| 26 | index 0000000..5e4233e |
| 27 | --- /dev/null |
| 28 | +++ b/internal/daemon/jobs_browse.go |
| 29 | @@ -0,0 +1,316 @@ |
| 30 | +package daemon |
| 31 | + |
| 32 | +// Long task browsing: the NIP-90 job requests the caller may see, each with |
| 33 | +// its newest feedback status and its result when one exists, and one |
| 34 | +// request with its feedback timeline and results. Both read through the |
| 35 | +// tenant gate with the caller's session, like the other browse methods. |
| 36 | + |
| 37 | +import ( |
| 38 | + "context" |
| 39 | + "encoding/json" |
| 40 | + "errors" |
| 41 | + "fmt" |
| 42 | + "strconv" |
| 43 | + "time" |
| 44 | + |
| 45 | + "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 46 | + "github.com/FelineStateMachine/tinyrelay/internal/relay" |
| 47 | + "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 48 | +) |
| 49 | + |
| 50 | +// jobScan bounds how many stored requests or answers one listing walks. |
| 51 | +const jobScan = 1000 |
| 52 | + |
| 53 | +var jobStates = []string{"open", "done", "all"} |
| 54 | + |
| 55 | +// jobRequestKinds lists every request kind, so one filter covers the range. |
| 56 | +var jobRequestKinds = func() []int { |
| 57 | + kinds := make([]int, 0, event.KIND_JOB_REQUEST_MAX-event.KIND_JOB_REQUEST_MIN+1) |
| 58 | + for kind := event.KIND_JOB_REQUEST_MIN; kind <= event.KIND_JOB_REQUEST_MAX; kind++ { |
| 59 | + kinds = append(kinds, kind) |
| 60 | + } |
| 61 | + return kinds |
| 62 | +}() |
| 63 | + |
| 64 | +type jobBrowseRequest struct { |
| 65 | + Limit int `json:"limit"` |
| 66 | + Cursor string `json:"cursor"` |
| 67 | + State string `json:"state"` |
| 68 | + Mine bool `json:"mine"` |
| 69 | + ID string `json:"id"` |
| 70 | + Event string `json:"event"` |
| 71 | +} |
| 72 | + |
| 73 | +// jobInput is one i tag of a request. |
| 74 | +type jobInput struct { |
| 75 | + Data string `json:"data"` |
| 76 | + Type string `json:"type"` |
| 77 | + Relay string `json:"relay,omitempty"` |
| 78 | + Marker string `json:"marker,omitempty"` |
| 79 | +} |
| 80 | + |
| 81 | +// jobParam is one param tag of a request. |
| 82 | +type jobParam struct { |
| 83 | + Key string `json:"key"` |
| 84 | + Value string `json:"value"` |
| 85 | +} |
| 86 | + |
| 87 | +// jobItem is one request as a browser sees it, settled against its answers. |
| 88 | +type jobItem struct { |
| 89 | + ID string `json:"id"` |
| 90 | + Kind int `json:"kind"` |
| 91 | + Requester string `json:"requester"` |
| 92 | + Providers []string `json:"providers"` |
| 93 | + Inputs []jobInput `json:"inputs"` |
| 94 | + Output string `json:"output,omitempty"` |
| 95 | + Params []jobParam `json:"params"` |
| 96 | + Bid string `json:"bid,omitempty"` |
| 97 | + Relays []string `json:"relays"` |
| 98 | + Encrypted bool `json:"encrypted,omitempty"` |
| 99 | + Content string `json:"content"` |
| 100 | + CreatedAt int64 `json:"created_at"` |
| 101 | + Expires int64 `json:"expires,omitempty"` |
| 102 | + State string `json:"state"` |
| 103 | + Status string `json:"status,omitempty"` |
| 104 | + Feedback *jobAnswer `json:"feedback,omitempty"` |
| 105 | + Result *jobAnswer `json:"result,omitempty"` |
| 106 | + Event event.Event `json:"event"` |
| 107 | +} |
| 108 | + |
| 109 | +// jobAnswer is one feedback or result event from a service provider. |
| 110 | +type jobAnswer struct { |
| 111 | + ID string `json:"id"` |
| 112 | + Kind int `json:"kind"` |
| 113 | + Provider string `json:"provider"` |
| 114 | + Status string `json:"status,omitempty"` |
| 115 | + Info string `json:"info,omitempty"` |
| 116 | + Amount string `json:"amount,omitempty"` |
| 117 | + Invoice string `json:"invoice,omitempty"` |
| 118 | + Content string `json:"content"` |
| 119 | + CreatedAt int64 `json:"created_at"` |
| 120 | +} |
| 121 | + |
| 122 | +func jobBrowseMethod(method string) bool { |
| 123 | + return method == "browsejobs" || method == "browsejob" |
| 124 | +} |
| 125 | + |
| 126 | +func (t *Tenant) executeJobs(ctx context.Context, actor, method string, params []json.RawMessage) (any, error) { |
| 127 | + if !t.Policy().Features.Jobs { |
| 128 | + return nil, errors.New("restricted: long tasks are switched off on this relay") |
| 129 | + } |
| 130 | + q := jobBrowseRequest{} |
| 131 | + if len(params) > 0 { |
| 132 | + if err := json.Unmarshal(params[0], &q); err != nil { |
| 133 | + return nil, fmt.Errorf("invalid: job parameters: %w", err) |
| 134 | + } |
| 135 | + } |
| 136 | + if q.Limit <= 0 || q.Limit > 100 { |
| 137 | + q.Limit = 50 |
| 138 | + } |
| 139 | + switch method { |
| 140 | + case "browsejobs": |
| 141 | + return t.browseJobs(ctx, actor, q) |
| 142 | + case "browsejob": |
| 143 | + if q.ID == "" { |
| 144 | + q.ID = q.Event |
| 145 | + } |
| 146 | + return t.browseJob(ctx, actor, q.ID) |
| 147 | + } |
| 148 | + return nil, errors.New("unsupported: browse operation") |
| 149 | +} |
| 150 | + |
| 151 | +// browseJobs lists the requests the caller may see, newest first. A request |
| 152 | +// is done once a result exists or its newest feedback reports success or an |
| 153 | +// error; every other request is open. Expired requests are not served. |
| 154 | +func (t *Tenant) browseJobs(ctx context.Context, actor string, q jobBrowseRequest) (any, error) { |
| 155 | + if q.State == "" { |
| 156 | + q.State = "all" |
| 157 | + } |
| 158 | + if !containsString(jobStates, q.State) { |
| 159 | + return nil, errors.New("invalid: job state must be open, done or all") |
| 160 | + } |
| 161 | + if q.Mine && actor == "" { |
| 162 | + return nil, errors.New("auth-required: sign in to list your own requests") |
| 163 | + } |
| 164 | + cursor, err := parseCollaborationCursor(q.Cursor) |
| 165 | + if err != nil { |
| 166 | + return nil, errors.New("invalid: job cursor") |
| 167 | + } |
| 168 | + session := browseSession(t, actor) |
| 169 | + now := time.Now().Unix() |
| 170 | + filter := event.Filter{Kinds: jobRequestKinds, Tags: map[string][]string{}} |
| 171 | + if q.Mine { |
| 172 | + filter.Authors = []string{actor} |
| 173 | + } |
| 174 | + items := make([]jobItem, 0, q.Limit) |
| 175 | + next := "" |
| 176 | + for scanned := 0; scanned < jobScan && next == ""; scanned += 100 { |
| 177 | + page, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: now, Access: storage.Access{PubKeys: session.PubKeys}, Limit: 100, Before: cursor}) |
| 178 | + if err != nil { |
| 179 | + return nil, err |
| 180 | + } |
| 181 | + visible := make([]jobItem, 0, len(page.Events)) |
| 182 | + for _, row := range page.Events { |
| 183 | + cursor = &storage.EventCursor{CreatedAt: row.CreatedAt, ID: row.ID} |
| 184 | + if t.gate.CanSee(ctx, row, session, nil) { |
| 185 | + visible = append(visible, jobItemFrom(row)) |
| 186 | + } |
| 187 | + } |
| 188 | + answers, err := t.jobAnswers(ctx, visible, session, now) |
| 189 | + if err != nil { |
| 190 | + return nil, err |
| 191 | + } |
| 192 | + for i := range visible { |
| 193 | + jobSettle(&visible[i], answers[visible[i].ID]) |
| 194 | + if q.State != "all" && visible[i].State != q.State { |
| 195 | + continue |
| 196 | + } |
| 197 | + if len(items) == q.Limit { |
| 198 | + next = jobCursor(items[len(items)-1]) |
| 199 | + break |
| 200 | + } |
| 201 | + items = append(items, visible[i]) |
| 202 | + } |
| 203 | + if !page.More || len(page.Events) == 0 { |
| 204 | + break |
| 205 | + } |
| 206 | + if next == "" && len(items) == q.Limit { |
| 207 | + next = jobCursor(items[len(items)-1]) |
| 208 | + } |
| 209 | + } |
| 210 | + return map[string]any{"items": items, "next_cursor": next}, nil |
| 211 | +} |
| 212 | + |
| 213 | +// browseJob returns one request with its feedback, oldest first, and its |
| 214 | +// results, newest first. |
| 215 | +func (t *Tenant) browseJob(ctx context.Context, actor, id string) (any, error) { |
| 216 | + if len(id) != 64 { |
| 217 | + return nil, errors.New("invalid: job id") |
| 218 | + } |
| 219 | + session := browseSession(t, actor) |
| 220 | + rows, err := t.Query(ctx, []event.Filter{{IDs: []string{id}, Kinds: jobRequestKinds, Limit: intPtr(1)}}, session) |
| 221 | + if err != nil { |
| 222 | + return nil, err |
| 223 | + } |
| 224 | + if len(rows) != 1 { |
| 225 | + return nil, errors.New("not found: job request") |
| 226 | + } |
| 227 | + item := jobItemFrom(rows[0]) |
| 228 | + answers, err := t.jobAnswers(ctx, []jobItem{item}, session, time.Now().Unix()) |
| 229 | + if err != nil { |
| 230 | + return nil, err |
| 231 | + } |
| 232 | + jobSettle(&item, answers[item.ID]) |
| 233 | + feedback, results := []jobAnswer{}, []jobAnswer{} |
| 234 | + for _, row := range answers[item.ID] { |
| 235 | + if row.Kind == event.KIND_JOB_FEEDBACK { |
| 236 | + feedback = append([]jobAnswer{jobAnswerFrom(row)}, feedback...) |
| 237 | + } else { |
| 238 | + results = append(results, jobAnswerFrom(row)) |
| 239 | + } |
| 240 | + } |
| 241 | + return map[string]any{"item": item, "feedback": feedback, "results": results}, nil |
| 242 | +} |
| 243 | + |
| 244 | +func jobItemFrom(e event.Event) jobItem { |
| 245 | + item := jobItem{ID: e.ID, Kind: e.Kind, Requester: e.PubKey, Providers: event.TagValues(e, "p"), Inputs: []jobInput{}, Params: []jobParam{}, Relays: []string{}, Output: event.Tag(e, "output"), Bid: event.Tag(e, "bid"), Content: e.Content, CreatedAt: e.CreatedAt, Expires: event.Expiration(e), State: "open", Event: e} |
| 246 | + for _, tag := range e.Tags { |
| 247 | + switch { |
| 248 | + case len(tag) >= 3 && tag[0] == "i": |
| 249 | + input := jobInput{Data: tag[1], Type: tag[2]} |
| 250 | + if len(tag) > 3 { |
| 251 | + input.Relay = tag[3] |
| 252 | + } |
| 253 | + if len(tag) > 4 { |
| 254 | + input.Marker = tag[4] |
| 255 | + } |
| 256 | + item.Inputs = append(item.Inputs, input) |
| 257 | + case len(tag) >= 3 && tag[0] == "param": |
| 258 | + item.Params = append(item.Params, jobParam{Key: tag[1], Value: tag[2]}) |
| 259 | + case len(tag) >= 2 && tag[0] == "relays": |
| 260 | + item.Relays = append(item.Relays, tag[1:]...) |
| 261 | + case len(tag) >= 1 && tag[0] == "encrypted": |
| 262 | + item.Encrypted = true |
| 263 | + } |
| 264 | + } |
| 265 | + return item |
| 266 | +} |
| 267 | + |
| 268 | +func jobAnswerFrom(row event.Event) jobAnswer { |
| 269 | + answer := jobAnswer{ID: row.ID, Kind: row.Kind, Provider: row.PubKey, Content: row.Content, CreatedAt: row.CreatedAt} |
| 270 | + for _, tag := range row.Tags { |
| 271 | + switch { |
| 272 | + case len(tag) >= 2 && tag[0] == "status" && answer.Status == "": |
| 273 | + answer.Status = tag[1] |
| 274 | + if len(tag) > 2 { |
| 275 | + answer.Info = tag[2] |
| 276 | + } |
| 277 | + case len(tag) >= 2 && tag[0] == "amount" && answer.Amount == "": |
| 278 | + answer.Amount = tag[1] |
| 279 | + if len(tag) > 2 { |
| 280 | + answer.Invoice = tag[2] |
| 281 | + } |
| 282 | + } |
| 283 | + } |
| 284 | + return answer |
| 285 | +} |
| 286 | + |
| 287 | +// jobAnswers finds the feedback and results naming the given requests in |
| 288 | +// one query, newest first, keeping only what the session may see. |
| 289 | +func (t *Tenant) jobAnswers(ctx context.Context, items []jobItem, session relay.Session, now int64) (map[string][]event.Event, error) { |
| 290 | + answers := make(map[string][]event.Event, len(items)) |
| 291 | + if len(items) == 0 { |
| 292 | + return answers, nil |
| 293 | + } |
| 294 | + ids := make([]string, 0, len(items)) |
| 295 | + kinds := []int{event.KIND_JOB_FEEDBACK} |
| 296 | + wanted := make(map[string]int, len(items)) |
| 297 | + for _, item := range items { |
| 298 | + ids = append(ids, item.ID) |
| 299 | + wanted[item.ID] = event.JobResultKind(item.Kind) |
| 300 | + if !containsInt(kinds, wanted[item.ID]) { |
| 301 | + kinds = append(kinds, wanted[item.ID]) |
| 302 | + } |
| 303 | + } |
| 304 | + rows, err := t.store.Query(ctx, event.Filter{Kinds: kinds, Tags: map[string][]string{"e": ids}}, storage.QueryOptions{Now: now, Access: storage.Access{PubKeys: session.PubKeys}, Limit: jobScan}) |
| 305 | + if err != nil { |
| 306 | + return nil, err |
| 307 | + } |
| 308 | + sortEventsNewestFirst(rows.Events) |
| 309 | + for _, row := range rows.Events { |
| 310 | + if !t.gate.CanSee(ctx, row, session, nil) { |
| 311 | + continue |
| 312 | + } |
| 313 | + for _, id := range event.TagValues(row, "e") { |
| 314 | + resultKind, ok := wanted[id] |
| 315 | + if ok && (row.Kind == event.KIND_JOB_FEEDBACK || row.Kind == resultKind) { |
| 316 | + answers[id] = append(answers[id], row) |
| 317 | + } |
| 318 | + } |
| 319 | + } |
| 320 | + return answers, nil |
| 321 | +} |
| 322 | + |
| 323 | +// jobSettle fills the newest feedback, the newest result and the state of |
| 324 | +// one request from its answers, which arrive newest first. |
| 325 | +func jobSettle(item *jobItem, rows []event.Event) { |
| 326 | + for _, row := range rows { |
| 327 | + answer := jobAnswerFrom(row) |
| 328 | + if row.Kind == event.KIND_JOB_FEEDBACK { |
| 329 | + if item.Feedback == nil { |
| 330 | + item.Feedback = &answer |
| 331 | + item.Status = answer.Status |
| 332 | + } |
| 333 | + } else if item.Result == nil { |
| 334 | + item.Result = &answer |
| 335 | + } |
| 336 | + } |
| 337 | + item.State = "open" |
| 338 | + if item.Result != nil || item.Status == "success" || item.Status == "error" { |
| 339 | + item.State = "done" |
| 340 | + } |
| 341 | +} |
| 342 | + |
| 343 | +func jobCursor(item jobItem) string { |
| 344 | + return strconv.FormatInt(item.CreatedAt, 10) + ":" + item.ID |
| 345 | +} |
| 346 | diff --git a/internal/daemon/jobs_test.go b/internal/daemon/jobs_test.go |
| 347 | new file mode 100644 |
| 348 | index 0000000..ec21113 |
| 349 | --- /dev/null |
| 350 | +++ b/internal/daemon/jobs_test.go |
| 351 | @@ -0,0 +1,164 @@ |
| 352 | +package daemon |
| 353 | + |
| 354 | +import ( |
| 355 | + "context" |
| 356 | + "encoding/json" |
| 357 | + "strconv" |
| 358 | + "strings" |
| 359 | + "testing" |
| 360 | + "time" |
| 361 | + |
| 362 | + "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 363 | +) |
| 364 | + |
| 365 | +// jobTenant returns a tenant with a member who requests jobs and an agent |
| 366 | +// granted to serve them. |
| 367 | +func jobTenant(t *testing.T) (*App, *Tenant, string, string) { |
| 368 | + t.Helper() |
| 369 | + app, tenant := testTenant(t) |
| 370 | + ctx := context.Background() |
| 371 | + owner := tenant.Policy().Owner |
| 372 | + member, _ := event.PublicKey(testMemberSecret) |
| 373 | + agent, _ := event.PublicKey(testAgentSecret) |
| 374 | + if _, err := tenant.Execute(ctx, owner, "setmember", []json.RawMessage{json.RawMessage(strconv.Quote(member)), json.RawMessage(`{"role":"member"}`)}); err != nil { |
| 375 | + t.Fatal(err) |
| 376 | + } |
| 377 | + now := time.Now().Unix() |
| 378 | + if err := publishAs(t, tenant, agentGrantEvent(t, agent, now, now+3600, []string{"jobs", "serve"})); err != nil { |
| 379 | + t.Fatalf("grant rejected: %v", err) |
| 380 | + } |
| 381 | + return app, tenant, member, agent |
| 382 | +} |
| 383 | + |
| 384 | +func TestBrowseJobsListsRequestsWithStatusAndResults(t *testing.T) { |
| 385 | + _, tenant, member, agent := jobTenant(t) |
| 386 | + ctx := context.Background() |
| 387 | + owner := tenant.Policy().Owner |
| 388 | + now := time.Now().Unix() |
| 389 | + |
| 390 | + open := wikiPublish(t, tenant, testMemberSecret, 5001, now-400, "", []string{"i", "hello", "text"}, []string{"i", strings.Repeat("d", 64), "event", "wss://relay.example", "context"}, []string{"output", "text/plain"}, []string{"bid", "1000"}, []string{"param", "lang", "es"}, []string{"relays", "wss://a.example", "wss://b.example"}, []string{"p", agent}) |
| 391 | + wikiPublish(t, tenant, testAgentSecret, 7000, now-350, "", []string{"e", open.ID}, []string{"p", member}, []string{"status", "payment-required"}, []string{"amount", "500", "lnbc1"}) |
| 392 | + working := wikiPublish(t, tenant, testAgentSecret, 7000, now-300, "", []string{"e", open.ID}, []string{"p", member}, []string{"status", "processing", "halfway"}) |
| 393 | + done := wikiPublish(t, tenant, testMemberSecret, 5002, now-390, "", []string{"i", "summarize", "text"}) |
| 394 | + wikiPublish(t, tenant, testAgentSecret, 7000, now-380, "", []string{"e", done.ID}, []string{"p", member}, []string{"status", "processing"}) |
| 395 | + result := wikiPublish(t, tenant, testAgentSecret, 6002, now-200, "the summary", []string{"e", done.ID}, []string{"p", member}, []string{"amount", "700"}, []string{"request", "{}"}) |
| 396 | + failed := wikiPublish(t, tenant, testOwnerSecret, 5001, now-300, "", []string{"i", "broken", "text"}) |
| 397 | + wikiPublish(t, tenant, testAgentSecret, 7000, now-250, "", []string{"e", failed.ID}, []string{"p", owner}, []string{"status", "error", "no such input"}) |
| 398 | + // An expired request is saved directly, as it would have been before it |
| 399 | + // lapsed, and is no longer served. |
| 400 | + expired := event.Event{Kind: 5001, CreatedAt: now - 7200, Content: "", Tags: [][]string{{"expiration", strconv.FormatInt(now-3600, 10)}}} |
| 401 | + if err := event.Sign(&expired, testMemberSecret); err != nil { |
| 402 | + t.Fatal(err) |
| 403 | + } |
| 404 | + if _, err := tenant.store.Save(ctx, expired, storageSave(tenant)); err != nil { |
| 405 | + t.Fatal(err) |
| 406 | + } |
| 407 | + note := wikiPublish(t, tenant, testMemberSecret, 1, now-100, "not a job") |
| 408 | + |
| 409 | + listed, err := approvalCall(t, tenant, member, "browsejobs", map[string]any{}) |
| 410 | + if err != nil { |
| 411 | + t.Fatal(err) |
| 412 | + } |
| 413 | + items := wikiItems(listed, "items") |
| 414 | + if len(items) != 3 || items[0]["id"] != failed.ID || items[1]["id"] != done.ID || items[2]["id"] != open.ID || listed["next_cursor"] != "" { |
| 415 | + t.Fatalf("listed %v", items) |
| 416 | + } |
| 417 | + item := approvalByID(items, open.ID) |
| 418 | + feedback, _ := item["feedback"].(map[string]any) |
| 419 | + if item["state"] != "open" || item["status"] != "processing" || item["kind"] != 5001.0 || item["requester"] != member || item["output"] != "text/plain" || item["bid"] != "1000" || item["result"] != nil || feedback == nil || feedback["id"] != working.ID || feedback["info"] != "halfway" || feedback["provider"] != agent { |
| 420 | + t.Fatalf("open item %v", item) |
| 421 | + } |
| 422 | + if inputs := item["inputs"].([]any); len(inputs) != 2 || inputs[0].(map[string]any)["type"] != "text" || inputs[1].(map[string]any)["relay"] != "wss://relay.example" || inputs[1].(map[string]any)["marker"] != "context" { |
| 423 | + t.Fatalf("inputs %v", item["inputs"]) |
| 424 | + } |
| 425 | + if params := item["params"].([]any); len(params) != 1 || params[0].(map[string]any)["key"] != "lang" || params[0].(map[string]any)["value"] != "es" { |
| 426 | + t.Fatalf("params %v", item["params"]) |
| 427 | + } |
| 428 | + if relays := item["relays"].([]any); len(relays) != 2 || relays[1] != "wss://b.example" { |
| 429 | + t.Fatalf("relays %v", item["relays"]) |
| 430 | + } |
| 431 | + if providers := item["providers"].([]any); len(providers) != 1 || providers[0] != agent { |
| 432 | + t.Fatalf("providers %v", item["providers"]) |
| 433 | + } |
| 434 | + item = approvalByID(items, done.ID) |
| 435 | + answer, _ := item["result"].(map[string]any) |
| 436 | + if item["state"] != "done" || item["status"] != "processing" || answer == nil || answer["id"] != result.ID || answer["kind"] != 6002.0 || answer["amount"] != "700" || answer["content"] != "the summary" { |
| 437 | + t.Fatalf("done item %v", item) |
| 438 | + } |
| 439 | + item = approvalByID(items, failed.ID) |
| 440 | + if item["state"] != "done" || item["status"] != "error" || item["result"] != nil { |
| 441 | + t.Fatalf("failed item %v", item) |
| 442 | + } |
| 443 | + |
| 444 | + for state, want := range map[string][]string{"open": {open.ID}, "done": {failed.ID, done.ID}} { |
| 445 | + filtered, err := approvalCall(t, tenant, member, "browsejobs", map[string]any{"state": state}) |
| 446 | + if err != nil { |
| 447 | + t.Fatal(err) |
| 448 | + } |
| 449 | + got := []string{} |
| 450 | + for _, item := range wikiItems(filtered, "items") { |
| 451 | + got = append(got, item["id"].(string)) |
| 452 | + } |
| 453 | + if strings.Join(got, ",") != strings.Join(want, ",") { |
| 454 | + t.Fatalf("state %s lists %v", state, got) |
| 455 | + } |
| 456 | + } |
| 457 | + mine, err := approvalCall(t, tenant, member, "browsejobs", map[string]any{"mine": true}) |
| 458 | + if err != nil { |
| 459 | + t.Fatal(err) |
| 460 | + } |
| 461 | + if items := wikiItems(mine, "items"); len(items) != 2 || approvalByID(items, failed.ID) != nil { |
| 462 | + t.Fatalf("mine %v", items) |
| 463 | + } |
| 464 | + if _, err := approvalCall(t, tenant, member, "browsejobs", map[string]any{"state": "bogus"}); err == nil || !strings.HasPrefix(err.Error(), "invalid:") { |
| 465 | + t.Fatalf("bogus state %v", err) |
| 466 | + } |
| 467 | + if _, err := approvalCall(t, tenant, "", "browsejobs", map[string]any{"mine": true}); err == nil || !strings.HasPrefix(err.Error(), "auth-required:") { |
| 468 | + t.Fatalf("guest mine %v", err) |
| 469 | + } |
| 470 | + // Guests see the requests on an open relay; pages follow the cursor. |
| 471 | + paged, err := approvalCall(t, tenant, "", "browsejobs", map[string]any{"limit": 2}) |
| 472 | + if err != nil { |
| 473 | + t.Fatal(err) |
| 474 | + } |
| 475 | + if items := wikiItems(paged, "items"); len(items) != 2 || paged["next_cursor"] == "" { |
| 476 | + t.Fatalf("paged %v", paged) |
| 477 | + } |
| 478 | + rest, err := approvalCall(t, tenant, "", "browsejobs", map[string]any{"limit": 2, "cursor": paged["next_cursor"]}) |
| 479 | + if err != nil { |
| 480 | + t.Fatal(err) |
| 481 | + } |
| 482 | + if items := wikiItems(rest, "items"); len(items) != 1 || items[0]["id"] != open.ID || rest["next_cursor"] != "" { |
| 483 | + t.Fatalf("second page %v", rest) |
| 484 | + } |
| 485 | + |
| 486 | + detail, err := approvalCall(t, tenant, member, "browsejob", map[string]any{"id": open.ID}) |
| 487 | + if err != nil { |
| 488 | + t.Fatal(err) |
| 489 | + } |
| 490 | + timeline := wikiItems(detail, "feedback") |
| 491 | + if item := detail["item"].(map[string]any); item["state"] != "open" || len(timeline) != 2 || timeline[0]["status"] != "payment-required" || timeline[0]["invoice"] != "lnbc1" || timeline[1]["id"] != working.ID || len(detail["results"].([]any)) != 0 { |
| 492 | + t.Fatalf("open detail %v", detail) |
| 493 | + } |
| 494 | + detail, err = approvalCall(t, tenant, member, "browsejob", map[string]any{"id": done.ID}) |
| 495 | + if err != nil { |
| 496 | + t.Fatal(err) |
| 497 | + } |
| 498 | + if results := wikiItems(detail, "results"); len(results) != 1 || results[0]["id"] != result.ID || len(wikiItems(detail, "feedback")) != 1 { |
| 499 | + t.Fatalf("done detail %v", detail) |
| 500 | + } |
| 501 | + for _, id := range []string{note.ID, expired.ID, strings.Repeat("f", 64)} { |
| 502 | + if _, err := approvalCall(t, tenant, member, "browsejob", map[string]any{"id": id}); err == nil || !strings.HasPrefix(err.Error(), "not found:") { |
| 503 | + t.Fatalf("browsejob %s: %v", id, err) |
| 504 | + } |
| 505 | + } |
| 506 | + // Switching long tasks off closes the browse surface too. |
| 507 | + policy := tenant.Policy() |
| 508 | + policy.Features.Jobs = false |
| 509 | + if err := tenant.applyPolicy(ctx, policy); err != nil { |
| 510 | + t.Fatal(err) |
| 511 | + } |
| 512 | + if _, err := approvalCall(t, tenant, member, "browsejobs", map[string]any{}); err == nil || !strings.Contains(err.Error(), "switched off") { |
| 513 | + t.Fatalf("browse with jobs off %v", err) |
| 514 | + } |
| 515 | +} |
| 516 | diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go |
| 517 | index 690c7df..f9e6e41 100644 |
| 518 | --- a/internal/telemetry/telemetry.go |
| 519 | +++ b/internal/telemetry/telemetry.go |
| 520 | @@ -236,7 +236,7 @@ func latencyBuckets() []float64 { |
| 521 | return []float64{.0005, .001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10} |
| 522 | } |
| 523 | |
| 524 | -var operationNames = map[string]struct{}{"http": {}, "websocket": {}, "publish": {}, "subscribe": {}, "query": {}, "count": {}, "fanout": {}, "delivery": {}, "commit": {}, "save": {}, "parse": {}, "authorize": {}, "upgrade": {}, "callback": {}, "delivery-discovery": {}, "notification-broadcast": {}, "browserepos": {}, "browserepo": {}, "browsefiles": {}, "browsefile": {}, "browsestatus": {}, "browseissues": {}, "browsepulls": {}, "browseissue": {}, "browsepull": {}, "browse-download": {}, "push": {}, "mcp": {}, "rooms": {}} |
| 525 | +var operationNames = map[string]struct{}{"http": {}, "websocket": {}, "publish": {}, "subscribe": {}, "query": {}, "count": {}, "fanout": {}, "delivery": {}, "commit": {}, "save": {}, "parse": {}, "authorize": {}, "upgrade": {}, "callback": {}, "delivery-discovery": {}, "notification-broadcast": {}, "browserepos": {}, "browserepo": {}, "browsefiles": {}, "browsefile": {}, "browsestatus": {}, "browseissues": {}, "browsepulls": {}, "browseissue": {}, "browsepull": {}, "browsejobs": {}, "browsejob": {}, "browse-download": {}, "push": {}, "mcp": {}, "rooms": {}} |
| 526 | var outcomeNames = map[string]struct{}{"success": {}, "ok": {}, "invalid": {}, "unauthorized": {}, "timeout": {}, "busy": {}, "closed": {}, "error": {}} |
| 527 | var queueNames = map[string]struct{}{"inbox": {}, "outbox": {}, "write": {}, "delivery": {}} |
| 528 | var workNames = map[string]struct{}{"delivery": {}, "push": {}, "replication": {}, "webhook": {}, "job": {}, "records-projection": {}, "site-mirror": {}, "git-metadata": {}, "catalog-owner": {}, "callback": {}, "delivery-discovery": {}, "notification-delivery": {}, "notification-broadcast": {}, "notification-push": {}, "notification": {}} |
| 529 | diff --git a/internal/webui/webmcp.go b/internal/webui/webmcp.go |
| 530 | index 93262cb..124f607 100644 |
| 531 | --- a/internal/webui/webmcp.go |
| 532 | +++ b/internal/webui/webmcp.go |
| 533 | @@ -26,7 +26,7 @@ func (a *App) webMCPQuery(writer http.ResponseWriter, request *http.Request) { |
| 534 | return |
| 535 | } |
| 536 | } |
| 537 | - allowed := map[string]bool{"browserepos": true, "browserepo": true, "browsefiles": true, "browsefile": true, "browsestatus": true, "browseissues": true, "browsepulls": true, "browseissue": true, "browsepull": true, "browsewiki": true, "browsewikipage": true, "browsewikimerge": true, "browserooms": true, "browseroom": true, "browsethread": true, "browseagent": true, "browseapprovals": true, "browseapproval": true} |
| 538 | + allowed := map[string]bool{"browserepos": true, "browserepo": true, "browsefiles": true, "browsefile": true, "browsestatus": true, "browseissues": true, "browsepulls": true, "browseissue": true, "browsepull": true, "browsewiki": true, "browsewikipage": true, "browsewikimerge": true, "browserooms": true, "browseroom": true, "browsethread": true, "browseagent": true, "browseapprovals": true, "browseapproval": true, "browsejobs": true, "browsejob": true} |
| 539 | if !allowed[method] { |
| 540 | writeJSON(writer, http.StatusBadRequest, map[string]string{"error": "unsupported WebMCP query"}) |
| 541 | return |
| 542 | diff --git a/internal/webui/webmcp.js b/internal/webui/webmcp.js |
| 543 | index 5928f4c..d083fe7 100644 |
| 544 | --- a/internal/webui/webmcp.js |
| 545 | +++ b/internal/webui/webmcp.js |
| 546 | @@ -104,6 +104,11 @@ |
| 547 | (input, signal) => query("browseapprovals", input, signal)); |
| 548 | register("tiny.read_approval", "Read one request for a decision by event id with every reaction and reply from the people asked.", |
| 549 | object({id: hash}, ["id"]), reads, (input, signal) => query("browseapproval", input, signal)); |
| 550 | + register("tiny.list_jobs", "List long task requests (NIP-90 job requests) visible to your account, each with its newest feedback status and its result when one exists. state narrows the list to open, done or all; mine lists only your own requests.", |
| 551 | + object({cursor: text, limit, state: {type: "string", enum: ["open", "done", "all"]}, mine: {type: "boolean"}}), reads, |
| 552 | + (input, signal) => query("browsejobs", input, signal)); |
| 553 | + register("tiny.read_job", "Read one long task request by event id with its feedback timeline and results.", |
| 554 | + object({id: hash}, ["id"]), reads, (input, signal) => query("browsejob", input, signal)); |
| 555 | const pageName = {type: "string", minLength: 1, maxLength: 512, description: "Page name or title; the relay normalizes it to the NIP-54 d tag."}; |
| 556 | register("tiny.list_wiki", "List wiki pages with their shown version, version count and open merge requests. Supports search by title or summary, an author filter and pagination.", |
| 557 | object({q: text, author: pubkey, cursor: text, limit}), reads, (input, signal) => query("browsewiki", input, signal)); |
| 558 | diff --git a/scripts/webmcp.test.mjs b/scripts/webmcp.test.mjs |
| 559 | index 6a2813e..199d513 100644 |
| 560 | --- a/scripts/webmcp.test.mjs |
| 561 | +++ b/scripts/webmcp.test.mjs |
| 562 | @@ -160,6 +160,32 @@ test("approval tools list, read and open requests for a decision", async () => { |
| 563 | assert.equal(opened.searchParams.get("answer"), null); |
| 564 | }); |
| 565 | |
| 566 | +test("job tools list and read long tasks through the session", async () => { |
| 567 | + const requests = []; |
| 568 | + const {tools} = await browser({path: "/r/work/tools", fetch: async url => { |
| 569 | + requests.push(new URL(url, "https://tiny.example")); |
| 570 | + return new Response(JSON.stringify({items: [{id: "a".repeat(64), kind: 5001, state: "open", status: "processing"}], next_cursor: "", |
| 571 | + item: {id: "b".repeat(64), kind: 5002, state: "done"}, feedback: [{status: "processing"}], results: [{kind: 6002}]})); |
| 572 | + }}); |
| 573 | + const list = tools.get("tiny.list_jobs"); |
| 574 | + assert.equal(list.annotations.readOnlyHint, true); |
| 575 | + assert.equal(list.inputSchema.properties.state.enum.join(","), "open,done,all"); |
| 576 | + assert.equal(list.inputSchema.properties.mine.type, "boolean"); |
| 577 | + const listed = await list.execute({state: "open", mine: true, limit: 5}); |
| 578 | + assert.equal(requests.at(-1).pathname, "/r/work/webmcp/query"); |
| 579 | + assert.equal(requests.at(-1).searchParams.get("method"), "browsejobs"); |
| 580 | + assert.deepEqual(JSON.parse(requests.at(-1).searchParams.get("params")), [{state: "open", mine: true, limit: 5}]); |
| 581 | + assert.equal(listed.structuredContent.result.items[0].status, "processing"); |
| 582 | + const read = tools.get("tiny.read_job"); |
| 583 | + assert.equal(read.annotations.readOnlyHint, true); |
| 584 | + assert.deepEqual([...read.inputSchema.required], ["id"]); |
| 585 | + const detail = await read.execute({id: "b".repeat(64)}); |
| 586 | + assert.equal(requests.at(-1).searchParams.get("method"), "browsejob"); |
| 587 | + assert.deepEqual(JSON.parse(requests.at(-1).searchParams.get("params")), [{id: "b".repeat(64)}]); |
| 588 | + assert.equal(detail.structuredContent.result.item.state, "done"); |
| 589 | + assert.equal(detail.structuredContent.result.results[0].kind, 6002); |
| 590 | +}); |
| 591 | + |
| 592 | test("link, files and notification tools reach the new surfaces", async () => { |
| 593 | const calls = []; |
| 594 | const {tools, sandbox} = await browser({path: "/r/work/tools", signedFetch: async (url, method, body) => { |