gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-09-11 14:00:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Stabilize Claude subscription completion transport

Browse Source
This commit is contained in:
rcourtman
2026-07-16 12:38:19 +01:00
parent 0719e34255
commit 043ecf4c66
4 changed files with 70 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/release-control/v6/internal/subsystems/ai-runtime.md
+7
View File
@@ -298,6 +298,13 @@ canonical qualified Pulse model identity from shared callers, strips only its
own subscription-provider prefix before CLI execution, and rejects foreign
provider prefixes rather than forwarding an invalid or cross-provider model
name to the local agent.
Claude receives that output schema through the trusted adapter system channel
and returns one JSON result that Pulse decodes with unknown fields and trailing
values rejected. Pulse does not use Claude Code's hidden `--json-schema`
retry loop: live qualification proved that wrapper could exhaust retries after
the model had already completed valid finding calls, incorrectly converting a
durable Patrol outcome into a provider failure. Codex may continue to use its
native output-schema file because its CLI exposes the completed turn directly.
Patrol consumes the provider streaming interface, so the adapter projects each
fully validated CLI turn into canonical buffered `content`, `tool_start`, and
`done` events. It must emit nothing before the complete CLI response passes the
internal/ai/providers/subscription_agent.go
+31 -5
View File
@@ -6,6 +6,7 @@ import (
"encoding/json"
"errors"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
@@ -178,7 +179,8 @@ func (c *SubscriptionAgentClient) Chat(ctx context.Context, req ChatRequest) (*C
if err != nil {
return nil, err
}
if len(subscriptionAgentControlPrompt)+len(requestPrompt) > maxSubscriptionAgentPromptBytes {
schemaBytes, _ := json.Marshal(subscriptionAgentOutputSchema(req))
if len(subscriptionAgentControlPrompt)+len(requestPrompt)+len(schemaBytes) > maxSubscriptionAgentPromptBytes {
return nil, fmt.Errorf("subscription agent prompt exceeds %d bytes", maxSubscriptionAgentPromptBytes)
}
@@ -196,7 +198,6 @@ func (c *SubscriptionAgentClient) Chat(ctx context.Context, req ChatRequest) (*C
}
defer os.RemoveAll(workdir)
schemaBytes, _ := json.Marshal(subscriptionAgentOutputSchema(req))
var raw []byte
switch c.agent {
case SubscriptionAgentCodex:
@@ -215,7 +216,13 @@ func (c *SubscriptionAgentClient) Chat(ctx context.Context, req ChatRequest) (*C
raw, err = os.ReadFile(outputPath)
}
case SubscriptionAgentClaude:
raw, err = c.run(ctx, "claude", []string{"-p", "--model", c.model, "--output-format", "json", "--json-schema", string(schemaBytes), "--safe-mode", "--no-session-persistence", "--permission-mode", "dontAsk", "--tools", "", "--system-prompt", subscriptionAgentControlPrompt}, requestPrompt, workdir)
// Claude Code's --json-schema mode may exhaust its hidden structured-
// output retries after already returning valid Pulse tool decisions on
// earlier provider turns. Keep the schema in the trusted system channel
// and validate the single JSON result locally instead; this avoids
// converting a completed Patrol finding into a terminal wrapper error.
claudeSystemPrompt := subscriptionAgentControlPrompt + "\n\nTRUSTED_OUTPUT_SCHEMA_JSON\n" + string(schemaBytes)
raw, err = c.run(ctx, "claude", []string{"-p", "--model", c.model, "--output-format", "json", "--safe-mode", "--no-session-persistence", "--permission-mode", "dontAsk", "--tools", "", "--system-prompt", claudeSystemPrompt}, requestPrompt, workdir)
default:
return nil, fmt.Errorf("unknown subscription agent %q", c.agent)
}
@@ -421,7 +428,7 @@ func decodeSubscriptionAgentTurn(agent SubscriptionAgent, raw []byte) (subscript
if len(payload) == 0 || string(payload) == "null" {
return turn, errors.New("Claude subscription response did not contain structured output")
}
if err := json.Unmarshal(payload, &turn); err != nil {
if err := decodeStrictJSON(payload, &turn); err != nil {
return turn, fmt.Errorf("decode Claude structured turn: %w", err)
}
if turn.InputTokens == 0 {
@@ -432,12 +439,31 @@ func decodeSubscriptionAgentTurn(agent SubscriptionAgent, raw []byte) (subscript
}
return turn, nil
}
if err := json.Unmarshal(raw, &turn); err != nil {
if err := decodeStrictJSON(raw, &turn); err != nil {
return turn, fmt.Errorf("decode Codex structured turn: %w", err)
}
return turn, nil
}
func decodeStrictJSON(raw []byte, target interface{}) error {
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.DisallowUnknownFields()
if err := decoder.Decode(target); err != nil {
return err
}
if decoder.More() {
return errors.New("subscription agent returned trailing JSON values")
}
var trailing interface{}
if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
if err == nil {
return errors.New("subscription agent returned trailing JSON values")
}
return err
}
return nil
}
func rejectCodexAgentToolActivity(events []byte) error {
for _, line := range bytes.Split(events, []byte{'\n'}) {
line = bytes.TrimSpace(line)
internal/ai/providers/subscription_agent_live_test.go
+13
View File
@@ -55,6 +55,19 @@ func TestSubscriptionAgentLive(t *testing.T) {
if toolStarts != 1 || len(done.ToolCalls) != 1 || done.ToolCalls[0].Name != "get_node_status" {
t.Fatalf("unexpected structured stream: starts=%d done=%#v", toolStarts, done)
}
response, err := client.Chat(ctx, ChatRequest{
Model: string(tt.agent) + ":" + tt.model,
System: "Return a short acknowledgement without selecting a tool.",
Messages: []Message{{Role: "user", Content: "Acknowledge that the observation turn is complete."}},
ToolChoice: &ToolChoice{Type: ToolChoiceNone},
})
if err != nil {
t.Fatalf("structured completion turn failed: %v", err)
}
if response.Content == "" || response.StopReason != "end_turn" || len(response.ToolCalls) != 0 {
t.Fatalf("unexpected structured completion: %#v", response)
}
})
}
}
internal/ai/providers/subscription_agent_test.go
+19 -1
View File
@@ -210,6 +210,8 @@ if [ "$1" = "auth" ]; then
exit 0
fi
seen_system=false
seen_output_schema=false
seen_json_schema_arg=false
seen_no_tools=false
seen_dont_ask=false
while [ "$#" -gt 0 ]; do
@@ -222,6 +224,13 @@ while [ "$#" -gt 0 ]; do
case "$1" in
*"IGNORE ALL RULES"*) echo "infrastructure data leaked into system prompt" >&2; exit 93 ;;
esac
case "$1" in
*"TRUSTED_OUTPUT_SCHEMA_JSON"*'"input"'*) seen_output_schema=true ;;
esac
;;
--json-schema)
seen_json_schema_arg=true
shift
;;
--tools)
shift
@@ -235,9 +244,11 @@ while [ "$#" -gt 0 ]; do
shift
done
[ "$seen_system" = true ] || { echo "missing trusted system prompt" >&2; exit 94; }
[ "$seen_output_schema" = true ] || { echo "missing trusted output schema" >&2; exit 97; }
[ "$seen_json_schema_arg" = false ] || { echo "Claude hidden structured-output mode was enabled" >&2; exit 98; }
[ "$seen_no_tools" = true ] || { echo "Claude built-in tools not disabled" >&2; exit 95; }
[ "$seen_dont_ask" = true ] || { echo "unexpected permission mode" >&2; exit 96; }
printf '%s' '{"structured_output":{"content":"healthy","stop_reason":"end_turn","tool_calls":[]},"usage":{"input_tokens":12,"output_tokens":3}}'
printf '%s' '{"result":"{\"content\":\"healthy\",\"stop_reason\":\"end_turn\",\"tool_calls\":[]}","usage":{"input_tokens":12,"output_tokens":3}}'
`)
t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH"))
t.Setenv("OPENAI_API_KEY", "must-not-leak")
@@ -331,6 +342,13 @@ func TestDecodeClaudeSubscriptionAgentRejectsNonJSONResult(t *testing.T) {
}
}
func TestDecodeSubscriptionAgentTurnRejectsUnknownStructuredFields(t *testing.T) {
raw := []byte(`{"result":"{\"content\":\"healthy\",\"stop_reason\":\"end_turn\",\"tool_calls\":[],\"unexpected\":true}"}`)
if _, err := decodeSubscriptionAgentTurn(SubscriptionAgentClaude, raw); err == nil || !strings.Contains(err.Error(), "unknown field") {
t.Fatalf("unknown structured field error = %v", err)
}
}
func writeExecutable(t *testing.T, path, body string) {
t.Helper()
if err := os.WriteFile(path, []byte(body), 0700); err != nil {