feat: select and prepare artifacts by exact compatibility profile

.github/workflows/ci.yml

@@ -71,6 +71,33 @@ jobs:
           ./onggi pull demo out
           diff -r src out
           ./onggi ls
+      - name: compatible profile preparation and warm cache
+        run: |
+          python3 - <<'PY'
+          import json
+          profile = dict(schema="onggi.profile.v1", model_source="sha256:"+"1"*64,
+              options="sha256:"+"2"*64, device="synthetic-ci", runtime="test-v1",
+              sdk="1", compiler="1", firmware="1", format="test", layout="test",
+              topology="single", tp=1, batch=4, sequence=256)
+          with open("profile.json", "w") as f:
+              json.dump(profile, f)
+          PY
+          export ONGGI_STORE="${ONGGI_TEST_S3/onggitest?/onggitest/cli?}"
+          ./onggi push --profile profile.json --ref compatible src
+          ./onggi prepare --request profile.json --cache node-cache --dir prepared compatible > prepared.json
+          diff -r src prepared
+          ./onggi prepare --request profile.json --cache node-cache --dir prepared-again compatible > prepared-again.json
+          diff -r src prepared-again
+          python3 - <<'PY'
+          import json
+          with open("prepared.json") as f:
+              assert json.load(f)["ready"] is True
+          with open("prepared-again.json") as f:
+              result = json.load(f)
+          assert result["ready"] is True
+          assert result["stats"]["read_bytes"] == 0
+          assert result["stats"]["cache_bytes"] > 0
+          PY
       - name: server log
         if: failure()
         run: tail -100 "$RUNNER_TEMP/srv/server.log"

README.ko.md

@@ -25,6 +25,11 @@ Onggi(옹기)는 서로 다른 바이트를 한 번만 저장하고, 복원한 
 
 ## 하는 일
 
+P3의 첫 단계로 [호환 프로필 선택·파일 사전 준비](docs/ko/compatible-prepare.md)를 지원합니다.
+`push --profile FILE`로 게시하고 `prepare --request FILE --dir DIR <ref>...`로
+호환 결과물을 골라 노드 캐시를 통해 검증된 파일을 완성한 뒤 게시합니다.
+파일 준비 단계이며 NPU 예열과 배치·확장 신호 연동은 후속 작업입니다.
+
 | | |
 |---|---|
 | **한 번만 저장** | 내용 기반 청크, 결과물별 매니페스트, 가중치는 배열 방식별로 한 벌 |

README.md

@@ -27,6 +27,11 @@ The Kubernetes Ref API, compile controller and fetches triggered by ai-rm/llm-d
 
 ## What it does
 
+The first P3 integration is [exact-profile artifact selection and preparation](docs/compatible-prepare.md):
+publish with `push --profile FILE`, then `prepare --request FILE --dir DIR <ref>...`
+chooses a compatible build and atomically prepares verified files through the node cache.
+This is file readiness; native NPU prewarming and placement/autoscaler integration remain planned.
+
 | | |
 |---|---|
 | **Store once** | content-defined chunks, manifests per artifact, weights stored once per weight layout |

cmd/onggi/main.go

@@ -23,6 +23,7 @@ const usage = `onggi — content-addressed store for model weights and compiled
 usage:
   onggi push    --store S [--ref NAME] <dir>     store a directory; only missing chunks are uploaded
   onggi pull    --store S <ref|sha256:...> <dir> restore an artifact, every chunk hash-checked
+  onggi prepare --store S --request FILE --dir DIR <ref>... select compatible files and prepare atomically
   onggi ls      --store S                        refs and store totals
   onggi inspect --store S <ref|sha256:...>       files of an artifact
   onggi diff    --store S <base> <ref>           what turning base into ref keeps, copies and fetches
@@ -55,6 +56,8 @@ func main() {
 		os.Exit(push(os.Args[2:], os.Stdout, os.Stderr))
 	case "pull":
 		os.Exit(pull(os.Args[2:], os.Stdout, os.Stderr))
+	case "prepare":
+		os.Exit(prepare(os.Args[2:], os.Stdout, os.Stderr))
 	case "ls":
 		os.Exit(ls(os.Args[2:], os.Stdout, os.Stderr))
 	case "inspect":

cmd/onggi/prepare.go

@@ -0,0 +1,164 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+
+	"github.com/ai-computing/onggi/pkg/backend"
+	bfs "github.com/ai-computing/onggi/pkg/backend/fs"
+	"github.com/ai-computing/onggi/pkg/digest"
+	"github.com/ai-computing/onggi/pkg/store"
+)
+
+func readProfileFile(name string) (store.Profile, error) {
+	f, err := os.Open(name)
+	if err != nil {
+		return store.Profile{}, err
+	}
+	defer f.Close()
+	return store.DecodeProfile(f)
+}
+
+type prepareReport struct {
+	Selection *store.Selection `json:"selection,omitempty"`
+	Ready     bool             `json:"ready"` // verified files on disk, not NPU readiness
+	Directory string           `json:"directory,omitempty"`
+	Stats     *store.PullStats `json:"stats,omitempty"`
+	Error     string           `json:"error,omitempty"`
+}
+
+func prepare(args []string, stdout, stderr io.Writer) int {
+	var report prepareReport
+	fail := func(code int, err error) int {
+		report.Error = err.Error()
+		fmt.Fprintln(stderr, "onggi:", err)
+		if writeJSON(stdout, stderr, report) != 0 {
+			return 1
+		}
+		return code
+	}
+	fs := flag.NewFlagSet("prepare", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	sf := addStoreFlags(fs)
+	request := fs.String("request", "", "exact compatibility profile JSON")
+	dest := fs.String("dir", "", "previously absent destination directory")
+	cacheDir := fs.String("cache", "", "node chunk cache directory")
+	peers := fs.String("peers", "", "comma-separated peer URLs")
+	workers := fs.Int("workers", 8, "ranges read at once")
+	planOnly := fs.Bool("plan-only", false, "select using metadata; do not fetch model chunks or create destination/cache")
+	if err := fs.Parse(args); err != nil {
+		return fail(2, err)
+	}
+	if fs.NArg() == 0 || *request == "" || (!*planOnly && *dest == "") || *workers < 1 {
+		return fail(2, errors.New("usage: onggi prepare --store S --request FILE [--plan-only | --dir DEST] <ref|digest>...; workers must be positive"))
+	}
+	want, err := readProfileFile(*request)
+	if err != nil {
+		return fail(2, err)
+	}
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	s, refs, err := sf.open(ctx)
+	if err != nil {
+		return fail(1, err)
+	}
+	release, err := s.Lease(ctx, "compatible prepare")
+	if err != nil {
+		return fail(1, err)
+	}
+	defer release()
+	candidates := make([]digest.Digest, 0, fs.NArg())
+	for _, input := range fs.Args() {
+		d, err := store.Resolve(ctx, refs, input)
+		if err != nil {
+			return fail(1, fmt.Errorf("candidate %q: %w", input, err))
+		}
+		candidates = append(candidates, d)
+	}
+	selected, err := s.Select(ctx, candidates, want)
+	report.Selection = &selected
+	if err != nil {
+		return fail(1, err)
+	}
+	if *planOnly {
+		return writeJSON(stdout, stderr, report)
+	}
+	d, err := digest.Parse(selected.Manifest)
+	if err != nil {
+		return fail(1, err)
+	}
+	o := store.PullOptions{Workers: *workers, Peers: peerClients(*peers)}
+	if *cacheDir != "" {
+		cb, err := bfs.Open(*cacheDir, backend.DurabilityNone)
+		if err == nil {
+			o.Cache, err = store.OpenWith(ctx, cb, store.Create{Chunker: s.Config.Chunker, Sub: s.Config.Sub, SubKinds: s.Config.SubKinds, PackSize: s.Config.PackSize})
+		}
+		if err != nil {
+			return fail(1, fmt.Errorf("node cache: %w", err))
+		}
+	}
+	st, err := prepareFiles(ctx, s, d, *dest, o)
+	if err != nil {
+		return fail(1, err)
+	}
+	report.Ready = true
+	report.Directory = *dest
+	report.Stats = &st
+	return writeJSON(stdout, stderr, report)
+}
+
+// prepareFiles publishes only a complete tree. It never re-resolves the chosen digest.
+func prepareFiles(ctx context.Context, s *store.Store, d digest.Digest, dest string, o store.PullOptions) (store.PullStats, error) {
+	var st store.PullStats
+	if err := ctx.Err(); err != nil {
+		return st, err
+	}
+	dest = filepath.Clean(dest)
+	parent := filepath.Dir(dest)
+	if err := os.MkdirAll(parent, 0755); err != nil {
+		return st, err
+	}
+	lock := dest + ".onggi-lock"
+	if err := os.Mkdir(lock, 0700); err != nil {
+		return st, fmt.Errorf("destination busy: %w", err)
+	}
+	defer os.Remove(lock) // never remove a lock this call did not create
+	if _, err := os.Lstat(dest); !errors.Is(err, os.ErrNotExist) {
+		if err == nil {
+			err = errors.New("destination already exists")
+		}
+		return st, err
+	}
+	tmp, err := os.MkdirTemp(parent, "."+filepath.Base(dest)+".onggi-")
+	if err != nil {
+		return st, err
+	}
+	defer os.RemoveAll(tmp)
+	st, err = s.Pull(ctx, d, tmp, o)
+	if err != nil {
+		return st, err
+	}
+	if err = ctx.Err(); err != nil {
+		return st, err
+	}
+	if err = os.Chmod(tmp, 0755); err != nil {
+		return st, err
+	}
+	if _, err = os.Lstat(dest); !errors.Is(err, os.ErrNotExist) {
+		if err == nil {
+			err = errors.New("destination appeared during preparation")
+		}
+		return st, err
+	}
+	if err = os.Rename(tmp, dest); err != nil {
+		return st, err
+	}
+	return st, nil
+}