fix(ingest): parse docmark's SSE-framed response, not bare JSON
Found via a real live-cluster end-to-end test (POST /ingest-path with a real
.docx against the deployed ingestion + docmark): 'decode docmark response:
invalid character e looking for beginning of value'. docmark's Streamable-HTTP
transport frames every response as SSE (event: message\r\ndata: {...}\r\n\r\n,
Content-Type: text/event-stream) -- stateless_http=True removes the need for
an initialize handshake/session ID, it does NOT change the wire framing to
bare JSON. My original client + its own test mock both assumed bare JSON,
so the unit tests passed while the real call failed.
sseDataPayload() extracts the data: line before JSON-unmarshaling; falls back
to the raw body if no SSE framing is present (forward-compatible). Test mock
(docmark_test.go) now emits the SAME framing the live server actually sends,
confirmed by directly probing docmark's live response before writing the fix.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -97,8 +97,13 @@ func extractViaDocmark(path string) (string, error) {
|
|||||||
return "", fmt.Errorf("docmark: HTTP %d: %s", resp.StatusCode, string(body))
|
return "", fmt.Errorf("docmark: HTTP %d: %s", resp.StatusCode, string(body))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
jsonBody := body
|
||||||
|
if data := sseDataPayload(body); data != nil {
|
||||||
|
jsonBody = data
|
||||||
|
}
|
||||||
|
|
||||||
var out docmarkResponse
|
var out docmarkResponse
|
||||||
if err := json.Unmarshal(body, &out); err != nil {
|
if err := json.Unmarshal(jsonBody, &out); err != nil {
|
||||||
return "", fmt.Errorf("decode docmark response: %w", err)
|
return "", fmt.Errorf("decode docmark response: %w", err)
|
||||||
}
|
}
|
||||||
if out.Error != nil {
|
if out.Error != nil {
|
||||||
@@ -114,6 +119,23 @@ func extractViaDocmark(path string) (string, error) {
|
|||||||
return text, nil
|
return text, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// sseDataPayload extracts the JSON payload from an SSE-framed response body
|
||||||
|
// ("event: message\r\ndata: {...}\r\n\r\n"). docmark's Streamable-HTTP
|
||||||
|
// transport frames every response this way (Content-Type: text/event-stream)
|
||||||
|
// regardless of stateless_http — that flag removes the session/initialize
|
||||||
|
// requirement, not the SSE wire framing. Returns nil if body isn't SSE-framed
|
||||||
|
// (e.g. a plain-JSON response, kept as a fallback for forward-compatibility).
|
||||||
|
func sseDataPayload(body []byte) []byte {
|
||||||
|
const prefix = "data: "
|
||||||
|
for _, line := range bytes.Split(body, []byte("\n")) {
|
||||||
|
line = bytes.TrimRight(line, "\r")
|
||||||
|
if bytes.HasPrefix(line, []byte(prefix)) {
|
||||||
|
return bytes.TrimPrefix(line, []byte(prefix))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// fileBase returns the final path segment (like filepath.Base, kept local to
|
// fileBase returns the final path segment (like filepath.Base, kept local to
|
||||||
// avoid importing path/filepath just for this one call).
|
// avoid importing path/filepath just for this one call).
|
||||||
func fileBase(path string) string {
|
func fileBase(path string) string {
|
||||||
|
|||||||
@@ -30,9 +30,17 @@ type mcpToolResult struct {
|
|||||||
} `json:"error,omitempty"`
|
} `json:"error,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// writeMCPResponse mirrors docmark's REAL response framing (empirically
|
||||||
|
// confirmed against the live server): Content-Type: text/event-stream,
|
||||||
|
// body is SSE-framed ("event: message\r\ndata: {...}\r\n\r\n"), not bare
|
||||||
|
// JSON -- inherent to MCP Streamable-HTTP, independent of stateless_http.
|
||||||
func writeMCPResponse(w http.ResponseWriter, body mcpToolResult) {
|
func writeMCPResponse(w http.ResponseWriter, body mcpToolResult) {
|
||||||
w.Header().Set("Content-Type", "application/json")
|
payload, _ := json.Marshal(body)
|
||||||
_ = json.NewEncoder(w).Encode(body)
|
w.Header().Set("Content-Type", "text/event-stream")
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
_, _ = w.Write([]byte("event: message\r\ndata: "))
|
||||||
|
_, _ = w.Write(payload)
|
||||||
|
_, _ = w.Write([]byte("\r\n\r\n"))
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestExtractViaDocmark_Success(t *testing.T) {
|
func TestExtractViaDocmark_Success(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user