xinmi/internal/application/service/wiki_failure_capture.go

package service

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"os"
	"path/filepath"
	"strings"
	"syscall"
	"time"
	"unicode/utf8"

	"github.com/Tencent/WeKnora/internal/logger"
	"github.com/Tencent/WeKnora/internal/types"
)

const wikiFailureCaptureEnv = "WEKNORA_WIKI_FAILURE_CAPTURE"
const wikiFailureCaptureVersion = "wiki-identity-failure-v1"

type wikiFailureCaptureScope struct {
	Directory       string    `json:"directory"`
	TenantID        uint64    `json:"tenant_id"`
	KnowledgeBaseID string    `json:"knowledge_base_id"`
	KnowledgeID     string    `json:"knowledge_id"`
	PendingOpID     int64     `json:"pending_op_id"`
	Attempt         int       `json:"attempt"`
	ModelID         string    `json:"model_id"`
	Phase           string    `json:"phase"`
	SourceHash      string    `json:"source_hash"`
	Release         string    `json:"release"`
	APIImage        string    `json:"api_image"`
	ExpiresAt       time.Time `json:"expires_at"`
}

type wikiIdentityFailureArtifact struct {
	Version         string                   `json:"version"`
	CapturedAt      time.Time                `json:"captured_at"`
	Status          string                   `json:"status"`
	TenantID        uint64                   `json:"tenant_id"`
	KnowledgeBaseID string                   `json:"knowledge_base_id"`
	KnowledgeID     string                   `json:"knowledge_id"`
	PendingOpID     int64                    `json:"pending_op_id"`
	Attempt         int                      `json:"attempt"`
	ModelID         string                   `json:"model_id"`
	Phase           string                   `json:"phase"`
	Release         string                   `json:"release"`
	APIImage        string                   `json:"api_image"`
	Validation      *wikiSourceIdentityError `json:"validation"`
	TemplateHash    string                   `json:"template_sha256"`
	InputHash       string                   `json:"template_input_hash"`
	RepairHash      string                   `json:"repair_instruction_sha256"`
	SourceHash      string                   `json:"source_sha256"`
	ResponseHash    string                   `json:"response_sha256"`
	SourceBytes     int                      `json:"source_bytes"`
	ResponseBytes   int                      `json:"response_bytes"`
	Source          string                   `json:"source,omitempty"`
	Response        string                   `json:"response,omitempty"`
}

func wikiFailureTextHash(text string) string {
	sum := sha256.Sum256([]byte(text))
	return hex.EncodeToString(sum[:])
}

// Recovery diagnostics are opt-in for one exact operation and source leaf.
// They never enter checkpoints, exports or public errors, and cannot retry work.
func captureWikiIdentityFailure(ctx context.Context, modelID, template string, input map[string]string, repair, raw string, original error) {
	outcome := retainWikiIdentityFailure(ctx, modelID, template, input, repair, raw, original)
	if outcome != "" {
		logger.Errorf(ctx, "wiki identity failure diagnostic: outcome=%s", outcome)
	}
}

func retainWikiIdentityFailure(ctx context.Context, modelID, template string, input map[string]string, repair, raw string, original error) string {
	setting := os.Getenv(wikiFailureCaptureEnv)
	if setting == "" {
		return ""
	}
	var scope wikiFailureCaptureScope
	decoder := json.NewDecoder(strings.NewReader(setting))
	decoder.DisallowUnknownFields()
	if len(setting) > 4096 || decoder.Decode(&scope) != nil || decoder.Decode(new(any)) != io.EOF ||
		scope.Directory == "" || !filepath.IsAbs(scope.Directory) || scope.TenantID == 0 ||
		scope.KnowledgeBaseID == "" || scope.KnowledgeID == "" || scope.PendingOpID <= 0 || scope.Attempt <= 0 ||
		scope.ModelID == "" || scope.Phase != "wiki_candidate_slug" || len(scope.SourceHash) != 64 ||
		scope.Release == "" || scope.APIImage == "" || !time.Now().Before(scope.ExpiresAt) {
		return "invalid_configuration"
	}
	guard := wikiRecoveryFromContext(ctx)
	tenant, ok := types.TenantIDFromContext(ctx)
	var validation *wikiSourceIdentityError
	if guard == nil || !ok || !errors.As(original, &validation) {
		return ""
	}
	row := guard.row
	if tenant != scope.TenantID || row.TenantID != tenant || row.KnowledgeBaseID != scope.KnowledgeBaseID ||
		row.KnowledgeID != scope.KnowledgeID || row.PendingOpID != scope.PendingOpID || row.AttemptNumber != scope.Attempt ||
		modelID != scope.ModelID || row.ModelID != modelID || row.ReviewedRelease != scope.Release ||
		row.ReviewedAPIImage != scope.APIImage || wikiPromptPurpose(template) != scope.Phase ||
		validation.SourceHash != scope.SourceHash || wikiCheckpointHash([]string{input["Content"]}) != scope.SourceHash {
		return ""
	}
	artifact := wikiIdentityFailureArtifact{
		Version: wikiFailureCaptureVersion, CapturedAt: time.Now().UTC(), Status: "complete",
		TenantID: tenant, KnowledgeBaseID: row.KnowledgeBaseID, KnowledgeID: row.KnowledgeID,
		PendingOpID: row.PendingOpID, Attempt: row.AttemptNumber, ModelID: modelID, Phase: scope.Phase,
		Release: row.ReviewedRelease, APIImage: row.ReviewedAPIImage, Validation: validation,
		TemplateHash: wikiFailureTextHash(template), InputHash: wikiCheckpointHash(input), RepairHash: wikiFailureTextHash(repair),
		SourceHash: wikiFailureTextHash(input["Content"]), ResponseHash: wikiFailureTextHash(raw),
		SourceBytes: len(input["Content"]), ResponseBytes: len(raw),
	}
	if !utf8.ValidString(input["Content"]) || !utf8.ValidString(raw) {
		artifact.Status = "invalid_utf8"
	} else if artifact.SourceBytes <= 256*1024 && artifact.ResponseBytes <= 1024*1024 {
		artifact.Source, artifact.Response = input["Content"], raw
	} else {
		artifact.Status = "size_limit"
	}
	encoded, err := json.Marshal(artifact)
	if err != nil {
		return "encoding_failed"
	}
	if len(encoded) > 2*1024*1024 {
		artifact.Source, artifact.Response, artifact.Status = "", "", "size_limit"
		encoded, _ = json.Marshal(artifact)
	}
	return writeWikiFailureArtifact(scope, encoded, artifact.Status)
}

func writeWikiFailureArtifact(scope wikiFailureCaptureScope, encoded []byte, status string) string {
	return writeWikiFailureArtifactWithFile(scope, encoded, status, func(root *os.Root, name string) (wikiFailureFile, error) {
		return root.OpenFile(name, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
	})
}

type wikiFailureFile interface {
	io.Writer
	Sync() error
	Close() error
}

func writeWikiFailureArtifactWithFile(scope wikiFailureCaptureScope, encoded []byte, status string, open func(*os.Root, string) (wikiFailureFile, error)) string {
	// The operator must provision a private, unserved mount. Resolve no symlinks;
	// os.Root keeps all later operations inside that opened directory.
	resolved, err := filepath.EvalSymlinks(scope.Directory)
	if err != nil || resolved != filepath.Clean(scope.Directory) {
		return "private_directory_unavailable"
	}
	if wikiFailureDirectoryIsServed(resolved) {
		return "private_directory_is_served"
	}
	root, err := os.OpenRoot(resolved)
	if err != nil {
		return "private_directory_unavailable"
	}
	defer root.Close()
	info, err := root.Stat(".")
	if err != nil || !info.IsDir() || info.Mode().Perm()&0077 != 0 {
		return "private_directory_permissions"
	}
	owner, ok := info.Sys().(*syscall.Stat_t)
	if !ok || uint64(owner.Uid) != uint64(os.Geteuid()) {
		return "private_directory_owner"
	}
	// Exclusive directory creation is a durable single-use claim across workers.
	// A failed write leaves the claim consumed, with no automatic capture retry.
	key := wikiCheckpointHash([]any{scope.TenantID, scope.KnowledgeBaseID, scope.KnowledgeID, scope.PendingOpID, scope.Attempt})
	if err := root.Mkdir(key, 0700); err != nil {
		if errors.Is(err, os.ErrExist) {
			return "already_claimed"
		}
		return "claim_failed"
	}
	if !syncWikiFailureDirectory(root, ".") {
		return "sync_failed"
	}
	part := filepath.Join(key, "failure.partial")
	file, err := open(root, part)
	if err != nil {
		return "write_failed"
	}
	written, writeErr := file.Write(encoded)
	syncErr := file.Sync()
	closeErr := file.Close()
	if written != len(encoded) || writeErr != nil || syncErr != nil || closeErr != nil {
		_ = root.Remove(part)
		return "write_failed"
	}
	if err := root.Rename(part, filepath.Join(key, "failure.json")); err != nil {
		_ = root.Remove(part)
		return "write_failed"
	}
	if !syncWikiFailureDirectory(root, key) {
		return "sync_failed"
	}
	return status
}

func wikiFailureDirectoryIsServed(directory string) bool {
	storage := strings.TrimSpace(os.Getenv("LOCAL_STORAGE_BASE_DIR"))
	if storage == "" {
		storage = "/data/files"
	}
	for _, base := range []string{storage, "/data/uploads", "/usr/share/nginx/html"} {
		base, err := filepath.Abs(base)
		if err != nil {
			return true
		}
		if resolved, err := filepath.EvalSymlinks(base); err == nil {
			base = resolved
		}
		relative, err := filepath.Rel(base, directory)
		if err != nil || (relative != ".." && !strings.HasPrefix(relative, ".."+string(filepath.Separator))) {
			return true
		}
	}
	return false
}

func syncWikiFailureDirectory(root *os.Root, name string) bool {
	directory, err := root.Open(name)
	if err != nil {
		return false
	}
	syncErr := directory.Sync()
	closeErr := directory.Close()
	return syncErr == nil && closeErr == nil
}