返回 DeepSeek-Reasonix
submission_identity.go
根目录 / internal / control / submission_identity.go
1 package control
2
3 import (
4 "context"
5 "crypto/sha256"
6 "encoding/hex"
7 "encoding/json"
8 "errors"
9 "log/slog"
10 "regexp"
11 "strings"
12 "sync"
13 "sync/atomic"
14
15 "reasonix/internal/agent"
16 "reasonix/internal/attachment"
17 "reasonix/internal/event"
18 "reasonix/internal/session"
19 )
20
21 // SubmissionRequest fingerprints the actual operation, never only its label.
22 type SubmissionRequest struct {
23 ID string `json:"-"`
24 HTTP bool `json:"http,omitempty"`
25 Input string `json:"input"`
26 Display string `json:"display,omitempty"`
27 Format string `json:"format,omitempty"`
28 Action string `json:"action,omitempty"`
29 RecoveryID string `json:"recoveryId,omitempty"`
30 Original string `json:"original,omitempty"`
31 Goal string `json:"goal,omitempty"`
32 ToolApprovalMode string `json:"toolApprovalMode,omitempty"`
33 Invocations []InvocationRequest `json:"invocations,omitempty"`
34 DraftIDs []string `json:"draftIds,omitempty"`
35 AttachmentDigests []string `json:"attachmentDigests,omitempty"`
36 Attachments []SubmissionAttachment `json:"attachments,omitempty"`
37 frozenSources map[string]*attachment.AttachmentRef
38 inheritedSources []attachment.Source
39 }
40
41 // SubmissionAttachment describes an ordered logical item, independently of
42 // its temporary credential. Reference reads still require session authority.
43 type SubmissionAttachment struct {
44 ClientAttachmentID string `json:"clientAttachmentId"`
45 DraftID string `json:"draftId,omitempty"`
46 Path string `json:"path,omitempty"`
47 Reference *attachment.AttachmentRef `json:"reference,omitempty"`
48 }
49
50 // ErrSubmissionNotAccepted is only attached to failures that precede durable
51 // admission. Transport cancellation and flush failures intentionally omit it:
52 // the caller must query the receipt before deciding whether a retry may run.
53 var ErrSubmissionNotAccepted = errors.New("submission not accepted")
54
55 type submissionIdentityState struct {
56 mu sync.Mutex
57 pending atomic.Pointer[pendingSubmissionAdmission]
58 reused atomic.Uint64
59 conflicts atomic.Uint64
60 unknown atomic.Uint64
61 }
62
63 // pendingSubmissionAdmission is the immutable fact persisted with turn/start.
64 // The execution path receives the same images through turnAdmission, so it does
65 // not need to consult controller-scoped temporary state.
66 type pendingSubmissionAdmission struct {
67 receipt session.SubmissionReceipt
68 imageInputs []attachment.ImageInput
69 durableCtx context.Context
70 }
71
72 type turnAdmission struct {
73 images preparedImageReferences
74 durableCtx context.Context
75 result *admissionResult
76 }
77
78 func newTurnAdmission(ctx context.Context, images preparedImageReferences) turnAdmission {
79 if ctx == nil {
80 ctx = context.Background()
81 }
82 return turnAdmission{images: images, durableCtx: ctx}
83 }
84
85 func (c *Controller) appendSubmissionEvent(events []session.Event, turnID string) []session.Event {
86 if pending := c.submissions.pending.Load(); pending != nil {
87 receipt := pending.receipt
88 receipt.TurnID = turnID
89 payload := struct {
90 session.SubmissionReceipt
91 ImageInputs []attachment.ImageInput `json:"imageInputs,omitempty"`
92 }{SubmissionReceipt: receipt, ImageInputs: pending.imageInputs}
93 data, _ := json.Marshal(payload)
94 return append(events, session.Event{Kind: "submission/accepted", Optional: true, Payload: data})
95 }
96 return events
97 }
98
99 func (c *Controller) trySubmissionAdmissionLock() func() {
100 if !c.submissions.mu.TryLock() {
101 return nil
102 }
103 return sync.OnceFunc(c.submissions.mu.Unlock)
104 }
105
106 func (c *Controller) releaseSubmissionAdmission() {
107 c.submissions.mu.Unlock()
108 // A synchronous queue dispatch may have deferred while submit owned the gate.
109 c.maybeDispatchInbox()
110 }
111
112 func submissionFingerprint(req SubmissionRequest) string {
113 data, _ := json.Marshal(req)
114 digest := sha256.Sum256(data)
115 return hex.EncodeToString(digest[:])
116 }
117
118 const submissionFingerprintVersion = 1
119
120 func canonicalSubmissionFingerprint(req SubmissionRequest) string {
121 req.AttachmentDigests = nil
122 for i, item := range req.Attachments {
123 transport := item.Path
124 if item.DraftID != "" {
125 transport = "draft:" + item.DraftID
126 }
127 if transport != "" {
128 req.Display = strings.ReplaceAll(req.Display, "("+transport+")", "(attachment:"+item.ClientAttachmentID+")")
129 req.Input = canonicalAttachmentToken(req.Input, transport, item.ClientAttachmentID)
130 req.Original = canonicalAttachmentToken(req.Original, transport, item.ClientAttachmentID)
131 }
132 if item.DraftID != "" {
133 req.DraftIDs = withoutDraftID(req.DraftIDs, item.DraftID)
134 }
135 if i == 0 {
136 req.Attachments = append([]SubmissionAttachment(nil), req.Attachments...)
137 }
138 req.Attachments[i] = SubmissionAttachment{ClientAttachmentID: item.ClientAttachmentID}
139 }
140 return submissionFingerprint(req)
141 }
142
143 // MatchesSubmissionReceipt validates a cold durable receipt without creating
144 // a Controller. Version zero keeps the exact pre-attachment fingerprint.
145 func MatchesSubmissionReceipt(req SubmissionRequest, receipt session.SubmissionReceipt) bool {
146 if req.ID != receipt.SubmissionID {
147 return false
148 }
149 switch receipt.FingerprintVersion {
150 case 0:
151 return receipt.Fingerprint == submissionFingerprint(req)
152 case submissionFingerprintVersion:
153 return receipt.Fingerprint == canonicalSubmissionFingerprint(req)
154 default:
155 return false
156 }
157 }
158
159 // LookupSubmission also detects accidental reuse of a key for different input.
160 func (c *Controller) LookupSubmission(req SubmissionRequest) (session.SubmissionReceipt, bool, error) {
161 return c.LookupSubmissionContext(c.attachmentContext(), req)
162 }
163
164 func (c *Controller) LookupSubmissionContext(ctx context.Context, req SubmissionRequest) (session.SubmissionReceipt, bool, error) {
165 store := c.sessionEventStore()
166 if store == nil || req.ID == "" {
167 return session.SubmissionReceipt{}, false, nil
168 }
169 receipt, ok := store.Submission(req.ID)
170 fingerprint := canonicalSubmissionFingerprint(req)
171 if receipt.FingerprintVersion == 0 {
172 fingerprint = submissionFingerprint(req)
173 }
174 if ok && receipt.FingerprintVersion > submissionFingerprintVersion {
175 return receipt, true, errors.New("submission receipt version is unsupported; automatic replay is disabled")
176 }
177 if ok && receipt.Fingerprint != fingerprint {
178 if receipt.FingerprintVersion == 0 {
179 return receipt, true, errors.New("legacy submission receipt cannot be verified; automatic replay is disabled")
180 }
181 c.submissions.conflicts.Add(1)
182 slog.Warn("submission conflict", "submissionId", req.ID)
183 return receipt, true, errors.New("submission identity conflicts with different input")
184 }
185 if ok {
186 c.submissions.reused.Add(1)
187 // The projection may already contain an asynchronously accepted batch.
188 // A retry is acknowledged only after its canonical journal is durable.
189 if _, err := store.Flush(ctx); err != nil {
190 c.submissions.unknown.Add(1)
191 return receipt, true, err
192 }
193 slog.Debug("submission already accepted", "submissionId", req.ID, "turnId", receipt.TurnID)
194 }
195 return receipt, ok, nil
196 }
197
198 // SubmitIdentified serializes identity checking with synchronous turn admission.
199 func (c *Controller) SubmitIdentified(req SubmissionRequest) (session.SubmissionReceipt, error) {
200 return c.SubmitIdentifiedContext(c.attachmentContext(), req)
201 }
202
203 func (c *Controller) SubmitIdentifiedContext(ctx context.Context, req SubmissionRequest) (session.SubmissionReceipt, error) {
204 if len(req.ID) > 256 || strings.ContainsAny(req.ID, "\x00\r\n") {
205 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, errors.New("invalid submission identity"))
206 }
207 return c.submitIdentifiedWithSetupContext(ctx, req, nil, func(admission turnAdmission) {
208 c.submitIdentifiedRequestLocked(req, admission)
209 })
210 }
211
212 func (c *Controller) submitIdentified(req SubmissionRequest, submit func()) (session.SubmissionReceipt, error) {
213 return c.submitIdentifiedWithSetup(req, nil, func(turnAdmission) { submit() })
214 }
215
216 // SubmitIdentifiedWithSetup validates and freezes explicit image attachments
217 // before setup mutates host-visible session state. Desktop uses this for the
218 // initial Goal transaction, whose Goal/profile changes must not survive a
219 // rejected image submission.
220 func (c *Controller) SubmitIdentifiedWithSetup(req SubmissionRequest, setup func() error) (session.SubmissionReceipt, error) {
221 return c.SubmitIdentifiedWithSetupContext(c.attachmentContext(), req, setup)
222 }
223
224 func (c *Controller) SubmitIdentifiedWithSetupContext(ctx context.Context, req SubmissionRequest, setup func() error) (session.SubmissionReceipt, error) {
225 if len(req.ID) > 256 || strings.ContainsAny(req.ID, "\x00\r\n") {
226 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, errors.New("invalid submission identity"))
227 }
228 return c.submitIdentifiedWithSetupContext(ctx, req, setup, func(admission turnAdmission) {
229 c.submitIdentifiedRequestLocked(req, admission)
230 })
231 }
232
233 func (c *Controller) submitIdentifiedRequestLocked(req SubmissionRequest, admission turnAdmission) {
234 switch {
235 case req.Action == ProtocolRecoveryAction:
236 c.submitProtocolRecoveryLocked(req.RecoveryID, req.Input, admission)
237 case req.Action == "delivery-recovery" || req.Action == FinalReadinessRecoveryAction:
238 c.submitFinalReadinessRecoveryLocked(req.Display, req.Input, admission)
239 case req.Action == "shell":
240 c.runShell(req.Input, admission)
241 case req.HTTP:
242 c.submitHTTPWithFormatLocked(req.Input, req.Display, req.Format, admission)
243 case len(req.Invocations) > 0:
244 c.submitInvocationsLocked(req.Input, req.Display, req.Invocations, admission)
245 default:
246 c.submitLocked(req.Input, req.Display, req.Original, admission)
247 }
248 }
249
250 func (c *Controller) submitIdentifiedWithSetup(req SubmissionRequest, setup func() error, submit func(turnAdmission)) (session.SubmissionReceipt, error) {
251 return c.submitIdentifiedWithSetupContext(c.attachmentContext(), req, setup, submit)
252 }
253
254 func (c *Controller) submitIdentifiedWithSetupContext(ctx context.Context, req SubmissionRequest, setup func() error, submit func(turnAdmission)) (session.SubmissionReceipt, error) {
255 prepared, err := c.PrepareSubmission(ctx, req)
256 if err != nil {
257 c.notice(err.Error())
258 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, err)
259 }
260 return c.acceptPreparedSubmission(ctx, prepared, setup, submit)
261 }
262
263 func (c *Controller) acceptPreparedSubmission(ctx context.Context, candidate *PreparedSubmission, setup func() error, submit func(turnAdmission)) (session.SubmissionReceipt, error) {
264 ctx, cancel := c.NewAttachmentOperationContext(ctx)
265 defer cancel()
266 c.submissions.mu.Lock()
267 defer c.releaseSubmissionAdmission()
268 req := candidate.request
269 if candidate.scope != c.attachmentScope() {
270 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, errors.New("attachment target changed; please retry"))
271 }
272 if receipt, ok, err := c.LookupSubmissionContext(ctx, req); ok || err != nil {
273 return receipt, err
274 }
275 prepared := candidate.images
276 if err := ctx.Err(); err != nil {
277 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, err)
278 }
279 store := c.sessionEventStore()
280 if req.ID != "" {
281 if store == nil {
282 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, errors.New("durable submission identity unavailable"))
283 }
284 if c.Running() {
285 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, ErrTurnRunning)
286 }
287 }
288 if setup != nil {
289 if err := setup(); err != nil {
290 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, err)
291 }
292 }
293 admission := newTurnAdmission(ctx, prepared)
294 if req.ID == "" {
295 result := admissionResult(-1)
296 admission.result = &result
297 submit(admission)
298 if result != turnStarted {
299 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, errors.New("submission was not admitted"))
300 }
301 return session.SubmissionReceipt{}, nil
302 }
303 receipt := &session.SubmissionReceipt{SessionID: store.ID(), SubmissionID: req.ID,
304 FingerprintVersion: submissionFingerprintVersion, Fingerprint: canonicalSubmissionFingerprint(req), MessageID: agent.NewMessageID(), AcceptedAttachmentDigests: strings.Join(attachmentDigests(prepared), ",")}
305 pending := &pendingSubmissionAdmission{
306 receipt: *receipt,
307 imageInputs: append([]attachment.ImageInput(nil), prepared.inputs...),
308 durableCtx: ctx,
309 }
310 c.submissions.pending.Store(pending)
311 defer c.submissions.pending.Store(nil)
312 c.SetTurnSubmissionID(req.ID)
313 submit(admission)
314 if err := c.flushSubmissionAdmission(ctx); err != nil {
315 c.submissions.unknown.Add(1)
316 return session.SubmissionReceipt{}, err
317 }
318 accepted, ok := store.Submission(req.ID)
319 if !ok {
320 return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, errors.New("submission was not durably admitted"))
321 }
322 return accepted, nil
323 }
324
325 func (c *Controller) submissionForTurn(turnID string) (session.SubmissionReceipt, bool) {
326 store := c.sessionEventStore()
327 if store == nil {
328 return session.SubmissionReceipt{}, false
329 }
330 return store.SubmissionForTurn(turnID)
331 }
332
333 func (c *Controller) flushSubmissionAdmission(ctx context.Context) error {
334 if c.submissions.pending.Load() == nil {
335 return nil
336 }
337 _, err := c.sessionEventStore().Flush(ctx)
338 return err
339 }
340
341 func (c *Controller) submissionAdmissionContext() context.Context {
342 if pending := c.submissions.pending.Load(); pending != nil && pending.durableCtx != nil {
343 return pending.durableCtx
344 }
345 return context.Background()
346 }
347
348 func (c *Controller) prepareSubmissionImages(req SubmissionRequest) (preparedImageReferences, []ImageReferenceFailure) {
349 return c.prepareSubmissionImagesContext(c.attachmentContext(), req)
350 }
351
352 func (c *Controller) prepareSubmissionImagesContext(ctx context.Context, req SubmissionRequest) (preparedImageReferences, []ImageReferenceFailure) {
353 ids := append([]string(nil), req.DraftIDs...)
354 ids = append(ids, draftIDsFromInput(req.Input)...)
355 seen := make(map[string]bool)
356 unique := ids[:0]
357 for _, id := range ids {
358 if !seen[id] {
359 seen[id] = true
360 unique = append(unique, id)
361 }
362 }
363 ids = unique
364 prepared := preparedImageReferences{byPath: map[string]string{}}
365 svc := c.attachmentService()
366 sources := append([]attachment.Source(nil), req.inheritedSources...)
367 structured, failures := c.structuredImageSources(ctx, req.Attachments)
368 if len(failures) > 0 {
369 return preparedImageReferences{}, failures
370 }
371 sources = append(sources, structured...)
372 if len(req.Attachments) > 0 {
373 filtered := ids[:0]
374 for _, id := range ids {
375 found := false
376 for _, item := range req.Attachments {
377 if item.DraftID == id {
378 found = true
379 break
380 }
381 }
382 if !found {
383 filtered = append(filtered, id)
384 }
385 }
386 ids = filtered
387 }
388 for _, id := range ids {
389 key := "draft:" + id
390 ref := req.frozenSources[key]
391 if ref == nil {
392 draft, ok := svc.Drafts().Lookup(c.attachmentScope(), id)
393 if !ok {
394 return preparedImageReferences{}, imageFailuresFromAttachment(attachment.Error{Code: attachment.CodeMissing, Message: "draft credential is not valid"})
395 }
396 ref = &draft.Ref
397 }
398 sources = append(sources, attachment.Source{Existing: ref, DisplayName: ref.DisplayName, Path: key})
399 }
400 // Structured attachments promise image understanding for this turn. Legacy
401 // @.reasonix/attachments paths are also frozen below, but a text-only model may
402 // retain them as tool-readable references without a vision fallback.
403 prepared.requiresImageUnderstanding = len(sources) > 0
404 for _, source := range c.explicitImageSources(req.Input) {
405 if frozen := req.frozenSources[normalizedImageReferencePath(source.Path)]; frozen != nil {
406 source.Existing = frozen
407 }
408 duplicate := false
409 for _, existing := range sources {
410 if existing.Path != "" && normalizedImageReferencePath(existing.Path) == normalizedImageReferencePath(source.Path) {
411 duplicate = true
412 break
413 }
414 }
415 if !duplicate {
416 sources = append(sources, source)
417 }
418 }
419 if len(sources) == 0 {
420 return prepared, nil
421 }
422 sources = c.appendOrdinaryImageSources(req.Input, sources)
423 batch, err := svc.PrepareBatch(ctx, sources)
424 if err != nil {
425 return preparedImageReferences{}, imageFailuresFromAttachment(err)
426 }
427 refs, err := svc.CommitBatch(ctx, batch)
428 if err != nil {
429 return preparedImageReferences{}, imageFailuresFromAttachment(err)
430 }
431 if err := c.rebindPreparedDrafts(req, ids); err != nil {
432 return preparedImageReferences{}, imageFailuresFromAttachment(err)
433 }
434 prepared.inputs = svc.InputsFromRefs(refs)
435 for _, ref := range refs {
436 prepared.ordered = append(prepared.ordered, ref.Content.Digest)
437 }
438 for i, source := range sources {
439 if source.Path != "" {
440 prepared.byPath[normalizedImageReferencePath(source.Path)] = refs[i].Content.Digest
441 }
442 }
443 for i, item := range req.Attachments {
444 prepared.byPath["attachment:"+item.ClientAttachmentID] = refs[len(req.inheritedSources)+i].Content.Digest
445 }
446 prepared.inputs = append(prepared.inputs, legacyRemoteImageInputs(req.Input)...)
447 return prepared, nil
448 }
449
450 var draftIDPattern = regexp.MustCompile(`^draft:([0-9a-fA-F]{32})$`)
451
452 func draftIDsFromInput(input string) []string {
453 var ids []string
454 seen := map[string]bool{}
455 for _, token := range parseRefTokens(input) {
456 match := draftIDPattern.FindStringSubmatch(token)
457 if len(match) != 2 {
458 continue
459 }
460 id := match[1]
461 if seen[id] {
462 continue
463 }
464 seen[id] = true
465 ids = append(ids, id)
466 }
467 return ids
468 }
469
470 func attachmentDigests(prepared preparedImageReferences) []string {
471 if len(prepared.inputs) == 0 {
472 return nil
473 }
474 out := make([]string, 0, len(prepared.inputs))
475 for _, in := range prepared.inputs {
476 if in.Kind == attachment.KindAttachment && in.Attachment != nil {
477 out = append(out, in.Attachment.Content.Digest)
478 }
479 }
480 return out
481 }
482
483 func (c *Controller) flushSubmissionStart(ctx context.Context, kind event.Kind) error {
484 if kind != event.TurnStarted {
485 return nil
486 }
487 return c.flushSubmissionAdmission(ctx)
488 }
489
490 // TurnIDForSubmission exposes the synchronous admission receipt without
491 // depending on whether the provider is still running when the desktop call returns.
492 func (c *Controller) TurnIDForSubmission(submissionID string) string {
493 if store := c.sessionEventStore(); store != nil {
494 if receipt, ok := store.Submission(submissionID); ok {
495 return receipt.TurnID
496 }
497 }
498 ledger := c.turnEventLedger()
499 if ledger == nil {
500 return ""
501 }
502 return ledger.TurnIDForSubmission(submissionID)
503 }
504
504 lines GO