| 1 | package main
|
| 2 |
|
| 3 | import (
|
| 4 | "bufio"
|
| 5 | "context"
|
| 6 | "encoding/base64"
|
| 7 | "encoding/json"
|
| 8 | "errors"
|
| 9 | "fmt"
|
| 10 | "io"
|
| 11 | "os/exec"
|
| 12 | )
|
| 13 |
|
| 14 | // protocolTransport drives the in-process `oak-engine` (a thin Rust worker built
|
| 15 | // on oakvcs-core) instead of the `oak` CLI. It builds correct oak objects with
|
| 16 | // core and sends the push HTTP directly — no per-push subprocess, no working
|
| 17 | // tree, no workdir lock — so it can sustain far higher concurrency and reveal
|
| 18 | // the server's per-repo push ceiling. This is the ForgeMark-comparable
|
| 19 | // (in-process, go-git-style) headline number.
|
| 20 | //
|
| 21 | // One engine process is spawned per agent and driven serially over line-
|
| 22 | // delimited JSON on stdin/stdout, mirroring the agent's single-goroutine use.
|
| 23 | type protocolTransport struct {
|
| 24 | engineBin string
|
| 25 | remote string
|
| 26 | owner string
|
| 27 | name string
|
| 28 | token string
|
| 29 | author string
|
| 30 | branch string
|
| 31 |
|
| 32 | cmd *exec.Cmd
|
| 33 | in io.WriteCloser
|
| 34 | out *bufio.Reader
|
| 35 | dead bool
|
| 36 | }
|
| 37 |
|
| 38 | func newProtocolTransport(engineBin, remote, owner, name, token, author, branch string) *protocolTransport {
|
| 39 | return &protocolTransport{
|
| 40 | engineBin: engineBin, remote: remote, owner: owner, name: name,
|
| 41 | token: token, author: author, branch: branch,
|
| 42 | }
|
| 43 | }
|
| 44 |
|
| 45 | type engineResp struct {
|
| 46 | OK bool `json:"ok"`
|
| 47 | CAS bool `json:"cas"`
|
| 48 | NewHead string `json:"new_head"`
|
| 49 | CloneMs float64 `json:"clone_ms"`
|
| 50 | Err *string `json:"err"`
|
| 51 | }
|
| 52 |
|
| 53 | // call sends one command and reads exactly one response line.
|
| 54 | func (t *protocolTransport) call(cmd map[string]any) (*engineResp, error) {
|
| 55 | if t.dead {
|
| 56 | return nil, errors.New("engine process is dead")
|
| 57 | }
|
| 58 | b, err := json.Marshal(cmd)
|
| 59 | if err != nil {
|
| 60 | return nil, err
|
| 61 | }
|
| 62 | if _, err := t.in.Write(append(b, '\n')); err != nil {
|
| 63 | t.dead = true
|
| 64 | return nil, fmt.Errorf("engine write: %w", err)
|
| 65 | }
|
| 66 | line, err := t.out.ReadBytes('\n')
|
| 67 | if err != nil {
|
| 68 | t.dead = true
|
| 69 | return nil, fmt.Errorf("engine read: %w", err)
|
| 70 | }
|
| 71 | var r engineResp
|
| 72 | if err := json.Unmarshal(line, &r); err != nil {
|
| 73 | return nil, fmt.Errorf("engine reply parse: %w (%q)", err, string(line))
|
| 74 | }
|
| 75 | return &r, nil
|
| 76 | }
|
| 77 |
|
| 78 | // respErr turns an engine reply into the harness's error convention: nil on
|
| 79 | // success; a CAS-marked error (classifyOak → contention) on 200+success:false;
|
| 80 | // the raw error otherwise.
|
| 81 | func respErr(r *engineResp, err error) error {
|
| 82 | if err != nil {
|
| 83 | return err
|
| 84 | }
|
| 85 | if r.OK {
|
| 86 | return nil
|
| 87 | }
|
| 88 | if r.CAS {
|
| 89 | return errors.New("push rejected: success:false (cas contention)")
|
| 90 | }
|
| 91 | if r.Err != nil {
|
| 92 | return errors.New(*r.Err)
|
| 93 | }
|
| 94 | return errors.New("push failed: engine reported not ok")
|
| 95 | }
|
| 96 |
|
| 97 | func (t *protocolTransport) Init(ctx context.Context) error {
|
| 98 | // The engine ignores ctx cancellation on setup; it's a fast local spawn.
|
| 99 | cmd := exec.Command(t.engineBin)
|
| 100 | stdin, err := cmd.StdinPipe()
|
| 101 | if err != nil {
|
| 102 | return err
|
| 103 | }
|
| 104 | stdout, err := cmd.StdoutPipe()
|
| 105 | if err != nil {
|
| 106 | return err
|
| 107 | }
|
| 108 | cmd.Stderr = engineStderr // shared diagnostic sink (set in main)
|
| 109 | if err := cmd.Start(); err != nil {
|
| 110 | return fmt.Errorf("start oak-engine: %w", err)
|
| 111 | }
|
| 112 | t.cmd, t.in, t.out = cmd, stdin, bufio.NewReaderSize(stdout, 1<<16)
|
| 113 |
|
| 114 | r, err := t.call(map[string]any{
|
| 115 | "op": "init", "remote": t.remote, "owner": t.owner, "name": t.name,
|
| 116 | "token": t.token, "author": t.author, "branch": t.branch,
|
| 117 | })
|
| 118 | if err != nil {
|
| 119 | return err
|
| 120 | }
|
| 121 | if !r.OK {
|
| 122 | if r.Err != nil {
|
| 123 | return fmt.Errorf("engine init: %s", *r.Err)
|
| 124 | }
|
| 125 | return errors.New("engine init failed")
|
| 126 | }
|
| 127 | return nil
|
| 128 | }
|
| 129 |
|
| 130 | func (t *protocolTransport) Push(ctx context.Context, files []FileSpec) error {
|
| 131 | wire := make([]map[string]string, len(files))
|
| 132 | for i, f := range files {
|
| 133 | wire[i] = map[string]string{
|
| 134 | "path": f.Path,
|
| 135 | "content_b64": base64.StdEncoding.EncodeToString(f.Content),
|
| 136 | }
|
| 137 | }
|
| 138 | r, err := t.call(map[string]any{"op": "push", "files": wire})
|
| 139 | return respErr(r, err)
|
| 140 | }
|
| 141 |
|
| 142 | // Session strategy is intentionally not supported on the protocol transport:
|
| 143 | // clone latency is server-dominated, so the cli transport is representative for
|
| 144 | // the read side, and keeping the engine to a single tracked branch keeps it
|
| 145 | // simple and correct. main.go rejects -strategy session -transport protocol.
|
| 146 | var errNoSession = errors.New("session strategy is cli-only; run -strategy session with -transport cli")
|
| 147 |
|
| 148 | func (t *protocolTransport) CloneSession(ctx context.Context) error { return errNoSession }
|
| 149 | func (t *protocolTransport) PushCheckpoint(ctx context.Context, session int, files []FileSpec) error {
|
| 150 | return errNoSession
|
| 151 | }
|
| 152 | func (t *protocolTransport) CleanupSession(session int) {}
|
| 153 |
|
| 154 | func (t *protocolTransport) Close() {
|
| 155 | if t.cmd == nil {
|
| 156 | return
|
| 157 | }
|
| 158 | if !t.dead && t.in != nil {
|
| 159 | _, _ = t.in.Write([]byte(`{"op":"quit"}` + "\n"))
|
| 160 | _ = t.in.Close()
|
| 161 | }
|
| 162 | _ = t.cmd.Wait()
|
| 163 | }
|