Add durable goals with scheduled and event-driven execution

PRIMITIVES.md

@@ -198,6 +198,31 @@ knowledge-graph layer and durable artifacts.
   `authzdecision/`, `scopeaudit/`, `relwritesaudit/`, `toolguardaudit/`,
   `toolchainaudit/`, `contentguardaudit/`, `metaagentaudit/`,
   `infoleakage{audit,decision,taint}/`
+- **Session schedules:** `pkg/agent/sessionschedule/` resolves bounded one-time,
+  interval, daily, and weekly schedules with timezone-aware quiet hours. It is
+  independent of goals and grants no execution authority. Goals currently supply
+  the approval, durable run ledger, and session activation adapter.
+- **Event sessions:** `pkg/agent/sessionevents/` and `pkg/memory/sessionevents/`
+  retain verified events and finite watches, with quiet hours, duplicate
+  suppression, and current source access. Registered consumers decide what a
+  watch can do; an event itself grants no authority.
+- **Private durable goals:** `pkg/agent/goals/` (lifecycle and ownership),
+  `pkg/memory/goals/` (transactional SQLite/PostgreSQL stores), `pkg/web/goals/`
+  (authenticated API and signed outbox),
+  `pkg/agent/tool/meta/capability/goals.go` (opt-in tools), and
+  `cmd/oap/internal/goalscmd/` (CLI). See `docs/operating/goals.md`.
+  `pkg/controllers/goals/` activates bounded private report sessions only after
+  signed human consent and an expiring execution grant. Active state alone
+  records intent; each session requires a fresh enforcing plan. Explicit finite
+  standing consent can authorize its private-delivery phase without another
+  human decision; the signed phase record names the original consent and run.
+  Scheduled collectors can publish explicitly approved private observations from
+  immutable run results. A registered source adapter retains source
+  dependencies; a durable outbox retries publication independently of runner
+  cleanup. These records authenticate the agent report, not independent external
+  facts. Separately approved discovery policies can suggest monitoring goals.
+  Each proposal needs exact human acceptance before a goal or watch exists;
+  finite question limits, expiry, and source access constrain those suggestions.
 - **Authorization of recall:** `pkg/memory/spicedbauthorizer/`
 - **Artifacts:** `pkg/platform/artifacts/` (versioning, labels, resolve),
   `pkg/platform/artifactstore/` (the `Store` interface + `blob/`, a gocloud.dev

cmd/oap/internal/auditcmd/truncation_test.go

@@ -29,3 +29,7 @@ func TestRequireCompleteAuditRead(t *testing.T) {
 		assert.NoError(t, requireCompleteAuditRead(memory.QueryResult{}))
 	})
 }
+
+func TestRequireCompleteAuditReadRejectsPartialExport(t *testing.T) {
+	require.ErrorContains(t, requireCompleteAuditRead(memory.QueryResult{Partial: true}), "incomplete audit export")
+}

cmd/oap/internal/auditcmd/verify.go

@@ -60,7 +60,7 @@ publisher's recorded chain head cannot be decoded — that head is what makes
 tail truncation detectable at all, so an undecodable one leaves the tail
 unverified. Unsigned entries (pre-provenance or written by an unsigned
 writer) are reported as a warning and do NOT, on their own, fail
-verification.`,
+verification. Requires administrative access to export the complete audit evidence.`,
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			return runAuditVerify(cmd, g, args[0], asJSON)
@@ -79,7 +79,7 @@ func runAuditVerify(cmd *cobra.Command, g *apcmd.Globals, sessionName string, as
 	if err != nil {
 		return err
 	}
-	conn, err := memclient.Connect(ctx, b, sessionName, "")
+	conn, err := memclient.ConnectAudit(ctx, b, sessionName)
 	if err != nil {
 		return err
 	}
@@ -93,10 +93,9 @@ func runAuditVerify(cmd *cobra.Command, g *apcmd.Globals, sessionName string, as
 	}
 	keys := buildKeySet(ctx, b, &sess, errOut)
 
-	// Every append-only entry in the scope. Empty Kinds = all kinds; we
-	// filter to append-only locally (kinds/all blank-import makes
-	// KindAppendOnly authoritative here).
-	res, err := conn.Client.Query(ctx, memory.Query{Scope: conn.Scope, Limit: auditReadLimit})
+	// Administrative export includes platform-only links that a session read
+	// would omit. Still filter locally to keep the verifier authoritative.
+	res, err := conn.Client.QueryAudit(ctx, conn.Scope)
 	if err != nil {
 		return fmt.Errorf("query memory: %w", err)
 	}
@@ -155,6 +154,9 @@ var errAuditTruncated = errors.New("audit verification failed: the session holds
 // the reader has no flag to raise, so the only honest answer is to say the
 // verification could not be performed.
 func requireCompleteAuditRead(res memory.QueryResult) error {
+	if res.Partial || len(res.DroppedPredicates) != 0 {
+		return fmt.Errorf("audit verification failed: incomplete audit export")
+	}
 	if !res.Truncated {
 		return nil
 	}

cmd/oap/internal/goalscmd/root.go

@@ -0,0 +1,131 @@
+// Package goalscmd implements session-authenticated management of private goals.
+package goalscmd
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/authzed/openagentprimitives/cmd/oap/internal/apcmd"
+	"github.com/authzed/openagentprimitives/cmd/oap/internal/memclient"
+	"github.com/authzed/openagentprimitives/pkg/agent/goals"
+	"github.com/spf13/cobra"
+)
+
+// NewCmd builds the oap goals subtree. Mutation request IDs are explicit so a
+// retry after a connection failure can return the original accepted revision.
+func NewCmd(g *apcmd.Globals) *cobra.Command {
+	root := &cobra.Command{Use: "goals", Short: "Manage private goals for a session's owner and agent class", Long: "Manage durable goals using an authenticated, non-delegated session. Execution requires separate bounded human consent. Responses are JSON."}
+	for _, op := range []string{"list", "get", "runs", "create", "update"} {
+		root.AddCommand(newOperation(g, op))
+	}
+	return root
+}
+
+func newOperation(g *apcmd.Globals, op string) *cobra.Command {
+	var req goals.Request
+	req.Operation = op
+	var due, timezone, title, outcome string
+	use := op + " <session>"
+	argc := 1
+	if op == "get" || op == "runs" || op == "update" {
+		use += " <goal-id>"
+		argc = 2
+	}
+	short := op + " goals"
+	if op == "runs" {
+		short = "List execution history for a goal"
+	}
+	cmd := &cobra.Command{Use: use, Short: short, Args: cobra.ExactArgs(argc)}
+	f := cmd.Flags()
+	switch op {
+	case "list", "runs":
+		if op == "list" {
+			f.StringVar((*string)(&req.List.State), "state", "", "Filter by state")
+		}
+		f.StringVar(&req.List.After, "after", "", "Continue from the previous response's next cursor")
+		f.IntVar(&req.List.Limit, "limit", 50, "Page size (1–100)")
+	case "create", "update":
+		f.StringVar(&title, "title", "", "Short goal title")
+		f.StringVar(&outcome, "outcome", "", "Desired outcome")
+		f.StringVar(&due, "due-at", "", "Intent only: RFC3339 due timestamp")
+		f.StringVar(&timezone, "timezone", "", "IANA timezone")
+		if op == "create" {
+			f.StringVar(&req.Create.RequestID, "request-id", "", "Stable idempotency key; reuse for identical retries")
+		} else {
+			f.StringVar(&req.Change.RequestID, "request-id", "", "Stable idempotency key; reuse for identical retries")
+			f.Int64Var(&req.Change.Revision, "revision", 0, "Expected current revision (required)")
+			f.StringVar(&req.Change.Action, "action", "revise", "revise, activate, pause, resume, cancel, or complete")
+			f.BoolVar(&req.Change.ClearDue, "clear-due", false, "Remove due timestamp")
+			req.Change.Result = &goals.Result{}
+			f.StringVar(&req.Change.Result.Summary, "summary", "", "Reported completion summary")
+			f.StringSliceVar(&req.Change.Result.Evidence, "evidence", nil, "Completion evidence references")
+		}
+	}
+	cmd.RunE = func(cmd *cobra.Command, args []string) error {
+		if argc == 2 {
+			req.ID = args[1]
+			req.Change.ID = args[1]
+		}
+		var dueAt *time.Time
+		if due != "" {
+			t, err := time.Parse(time.RFC3339, due)
+			if err != nil {
+				return fmt.Errorf("--due-at: %w", err)
+			}
+			dueAt = &t
+		}
+		if op == "create" {
+			if req.Create.RequestID == "" || title == "" || outcome == "" {
+				return fmt.Errorf("--request-id, --title, and --outcome are required")
+			}
+			req.Create.Title = title
+			req.Create.Outcome = outcome
+			req.Create.DueAt = dueAt
+			req.Create.Timezone = timezone
+		}
+		if op == "update" {
+			if req.Change.RequestID == "" || req.Change.Revision < 1 {
+				return fmt.Errorf("--request-id and --revision are required")
+			}
+			if f.Changed("title") {
+				req.Change.Title = &title
+			}
+			if f.Changed("outcome") {
+				req.Change.Outcome = &outcome
+			}
+			if f.Changed("timezone") {
+				req.Change.Timezone = &timezone
+			}
+			req.Change.DueAt = dueAt
+			if !f.Changed("summary") && !f.Changed("evidence") {
+				req.Change.Result = nil
+			}
+		}
+		b, err := g.Bundle()
+		if err != nil {
+			return err
+		}
+		conn, err := memclient.Connect(cmd.Context(), b, args[0], "")
+		if err != nil {
+			return err
+		}
+		defer conn.Close()
+		if op == "create" || op == "update" {
+			// Resolve the domain through the server; local flags never select an owner.
+			res, err := conn.Client.Goals(cmd.Context(), b.Namespace, args[0], goals.Request{Operation: "list", List: goals.ListRequest{Limit: 1}})
+			if err != nil {
+				return err
+			}
+			req.Resource = res.Resource
+		}
+		res, err := conn.Client.Goals(cmd.Context(), b.Namespace, args[0], req)
+		if err != nil {
+			return err
+		}
+		enc := json.NewEncoder(cmd.OutOrStdout())
+		enc.SetIndent("", "  ")
+		return enc.Encode(res)
+	}
+	return cmd
+}

cmd/oap/internal/goalscmd/root_test.go

@@ -0,0 +1,22 @@
+package goalscmd
+
+import (
+	"testing"
+
+	"github.com/authzed/openagentprimitives/cmd/oap/internal/apcmd"
+	"github.com/stretchr/testify/require"
+)
+
+func TestInvalidGoalCLIRequestRefusedBeforeConnecting(t *testing.T) {
+	for _, args := range [][]string{
+		{"create", "session", "--title", "Agenda", "--outcome", "Prepare"},
+		{"update", "session", "goal-id", "--request-id", "retry"},
+		{"create", "session", "--request-id", "retry", "--title", "Agenda", "--outcome", "Prepare", "--due-at", "tomorrow"},
+	} {
+		cmd := NewCmd(&apcmd.Globals{})
+		cmd.SetArgs(args)
+		err := cmd.Execute()
+		require.Error(t, err)
+		require.NotContains(t, err.Error(), "kubeconfig")
+	}
+}

cmd/oap/internal/installcmd/install.go

@@ -3164,22 +3164,24 @@ func ensureWebhookTLSWithClient(ctx context.Context, c client.Client) ([]byte, e
 // silently skipped rather than failing install — the patch will succeed once
 // the manifest bundle is applied.
 func patchWebhookVWCBundle(ctx context.Context, c client.Client, caPEM []byte) error {
-	var vwc admissionregistrationv1.ValidatingWebhookConfiguration
-	if err := c.Get(ctx, types.NamespacedName{Name: "spicebox-settings"}, &vwc); err != nil {
-		if apierrors.IsNotFound(err) {
-			// VWC not yet created (Task 6 lands it in a later unit). Skip the
-			// caBundle patch — install must not fail before the manifest bundle
-			// includes the VWC. The patch will succeed on the next oap install
-			// after the manifest is applied.
-			return nil
+	for _, configurationName := range []string{"spicebox-settings", "spicebox-goal-execution"} {
+		var vwc admissionregistrationv1.ValidatingWebhookConfiguration
+		if err := c.Get(ctx, types.NamespacedName{Name: configurationName}, &vwc); err != nil {
+			if apierrors.IsNotFound(err) {
+				// VWC not yet created (Task 6 lands it in a later unit). Skip the
+				// caBundle patch — install must not fail before the manifest bundle
+				// includes the VWC. The patch will succeed on the next oap install
+				// after the manifest is applied.
+				continue
+			}
+			return fmt.Errorf("get ValidatingWebhookConfiguration %s: %w", configurationName, err)
+		}
+		for i := range vwc.Webhooks {
+			vwc.Webhooks[i].ClientConfig.CABundle = caPEM
+		}
+		if err := c.Update(ctx, &vwc); err != nil {
+			return fmt.Errorf("update ValidatingWebhookConfiguration %s caBundle: %w", configurationName, err)
 		}
-		return fmt.Errorf("get ValidatingWebhookConfiguration spicebox-settings: %w", err)
-	}
-	for i := range vwc.Webhooks {
-		vwc.Webhooks[i].ClientConfig.CABundle = caPEM
-	}
-	if err := c.Update(ctx, &vwc); err != nil {
-		return fmt.Errorf("update ValidatingWebhookConfiguration spicebox-settings caBundle: %w", err)
 	}
 	return nil
 }
@@ -3237,6 +3239,7 @@ var webdNATSGrant = apnats.UserGrant{
 		"ap.session.*.*.in.view_message",
 		"ap.session.*.*.in.interrupt_request",
 		"ap.session.*.*.in.interaction_decision",
+		"ap.component.interaction_decision.*.*",
 		"ap.session.*.*.in.resurface_request",
 		"ap.session.*.*.in.app_tool_call",
 		// ui_data_binding is the agent-UI's own data path: pkg/web/webui/agentui's

cmd/oap/internal/installcmd/nats_grant_test.go

@@ -56,6 +56,7 @@ func TestWebdNATSGrantPublishSurface(t *testing.T) {
 
 	// The inbound subjects webd genuinely publishes.
 	permitted := map[string]string{
+		"ap.component.interaction_decision.ns1.sess1":  "client-hosted verified human decisions outside runner grants",
 		"ap.session.ns1.sess1.in.view_message":         "chat SubmitUserMessage + /interact user_message|annotation_batch|mcp_ui_action → RequestViewMessage",
 		"ap.session.ns1.sess1.in.interrupt_request":    "chat SubmitInterrupt → clienthosted.SubmitInterrupt",
 		"ap.session.ns1.sess1.in.interaction_decision": "chat SubmitDecision → clienthosted.SubmitInteractionDecision (the browser approve/deny button)",

cmd/oap/internal/memclient/memclient.go

@@ -32,9 +32,31 @@ func (c *Conn) Close() {
 // BaseURL is the port-forwarded operator base URL (no trailing slash).
 func (c *Conn) BaseURL() string { return c.pf.URL() }
 
-// Token is the per-session memory bearer token.
+// Token is the bearer used by this connection.
 func (c *Conn) Token() string { return c.token }
 
+// ConnectAudit uses the operator's administrative credential for a complete
+// audit export. Reading the Secret requires Kubernetes administrative access;
+// no agent/session credential is elevated or reused.
+func ConnectAudit(ctx context.Context, b *kube.Bundle, sessionName string) (*Conn, error) {
+	var sec corev1.Secret
+	if err := b.Controller.Get(ctx, client.ObjectKey{Namespace: "agentprimitives-system", Name: "spicebox-operator-debug-token"}, &sec); err != nil {
+		return nil, fmt.Errorf("read administrative audit credential: %w", err)
+	}
+	token := string(sec.Data["token"])
+	if token == "" {
+		return nil, fmt.Errorf("administrative audit credential is empty")
+	}
+	pf, err := portforward.New(b.REST, "agentprimitives-system", "spicebox-operator", "app.kubernetes.io/name=spicebox-operator", 8082, 0)
+	if err != nil {
+		return nil, err
+	}
+	if err := pf.Start(ctx, io.Discard); err != nil {
+		return nil, fmt.Errorf("audit port-forward: %w", err)
+	}
+	return &Conn{Client: httpclient.New(pf.URL(), token), Scope: memory.Scope{Kind: "session", ID: b.Namespace + "/" + sessionName}, pf: pf, token: token}, nil
+}
+
 // Connect resolves the memory bearer token from the session's K8s Secret,
 // opens a port-forward to the operator, and returns a connected Conn.
 // backend may be "" (default), "primary", or "secondary" to override

cmd/oap/internal/portforward/portforward.go

@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"net/url"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -83,7 +84,10 @@ func (p *PortForwarder) Start(ctx context.Context, out io.Writer) error {
 	if len(pods.Items) == 0 {
 		return fmt.Errorf("no pods in %s matching %q (service %q)", p.namespace, p.selector, p.service)
 	}
-	pod := pods.Items[0]
+	pod := readyPod(pods.Items)
+	if pod == nil {
+		return fmt.Errorf("no ready running pods in %s matching %q (service %q)", p.namespace, p.selector, p.service)
+	}
 
 	roundTripper, upgrader, err := spdy.RoundTripperFor(p.cfg)
 	if err != nil {
@@ -139,6 +143,23 @@ func (p *PortForwarder) Start(ctx context.Context, out io.Writer) error {
 	return nil
 }
 
+// Retained evicted pods and terminating rollout replicas can precede the live
+// service pod in a list. Forward only to a replica that can serve requests.
+func readyPod(pods []corev1.Pod) *corev1.Pod {
+	for i := range pods {
+		pod := &pods[i]
+		if pod.DeletionTimestamp != nil || pod.Status.Phase != corev1.PodRunning {
+			continue
+		}
+		for _, condition := range pod.Status.Conditions {
+			if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue {
+				return pod
+			}
+		}
+	}
+	return nil
+}
+
 // URL returns the localhost HTTP URL for the forwarded port. Convenience for
 // HTTP callers — gRPC callers should read LocalPort and compose their own
 // dial string.