feat(plugin): enqueue resource mutations

VERSION

@@ -1 +1 @@
-0.1.44
+0.1.45

docs/plugin.md

@@ -510,8 +510,8 @@ Resource type 使用 BatonSession 内的所有权注册表。Baton 启动时先
 
 ### 2.2 Resource 与 Controller
 
-Baton-owned Resource 更新、Resource 创建或 `spec` 更新、启动恢复、外部观察、Controller cron Source
-或动态计时到期都只表示
+Baton-owned Resource 更新、Plugin Resource 创建或有效的 `status` 更新、后续 `spec` 更新、启动恢复、
+外部观察、Controller cron Source 或动态计时到期都只表示
 “某个对象可能需要重新检查”。Baton 将同一对象的重复触发合并成 reconcile key：
 
 ```text
@@ -822,7 +822,8 @@ Proposal 重建。Baton-owned Resource 不在 `plugins/` 下另存副本。
    整体回滚，解绑和退出统一关闭。
 2. 已建立 Resource 通用信封与存储、同 key 不并发的 reconcile queue、持久 Proposal，
    `requeueAfter` due time 与 Controller cron Sources；三种唤醒都进入同一
-   keyed queue，运行数据全部归当前 BatonSession。
+   keyed queue；Plugin Resource 创建和有效的 status 更新也自动进入该队列，运行数据全部归当前
+   BatonSession。
 3. 已建立 `_baton_turn_summary` → `baton.dev/v1alpha1, Kind=Turn` 的只读 Baton-owned Resource，
    `registerController` 复用同一 queue、退避和 Proposal 管线；启动 replay 与 live append
    使用同一资源 key。

src/plugin/manager.ts

@@ -53,7 +53,10 @@ import {
   PluginResourceStore,
   resourceTypeKey,
 } from "./resource.ts";
-import { createResourceClient } from "./resource-client.ts";
+import {
+  createResourceClient,
+  type ResourceClientChange,
+} from "./resource-client.ts";
 import {
   type BoardItem,
   presentBoardSource,
@@ -503,7 +506,7 @@ export class Manager {
           session: this.instances.session,
           pluginInstanceId: instance.pluginInstanceId,
         }),
-        () => this.notifyBoardChanged(),
+        (change) => this.handlePluginResourceChange(change),
         (resourceType) =>
           this.claimResourceTypeForCreate(instance.pluginId, resourceType),
       ),
@@ -1168,6 +1171,34 @@ export class Manager {
     }
   }
 
+  private handlePluginResourceChange(change: ResourceClientChange): void {
+    this.notifyBoardChanged();
+    if (this.closed || change.kind === "deleted") return;
+
+    const key = Object.freeze({
+      batonSessionId: this.proposals.batonSessionId,
+      pluginInstanceId: change.resource.metadata.namespace,
+      resourceApiVersion: change.resource.apiVersion,
+      resourceKind: change.resource.kind,
+      resourceId: change.resource.metadata.name,
+    });
+    const scopeId = reconcileScopeId(key);
+    // Source discovery enqueues the complete Resource set when it finishes.
+    // Suppressing the immediate signal here avoids reconciling a newly discovered key twice.
+    if (this.activeSourceScopes.has(scopeId)) return;
+
+    const controller = this.controllers.get(scopeId);
+    if (
+      !controller ||
+      this.suspendedControllers.has(scopeId)
+    ) {
+      return;
+    }
+    void controller.enqueue(key).catch(() => {
+      // Reconcile failures use the Controller retry path; close races need no extra reaction.
+    });
+  }
+
   private notifyToast(
     pluginInstanceId: string,
     message: ToastMessage,

src/plugin/resource-client.ts

@@ -8,6 +8,11 @@ import { PluginResourceStore } from "./resource.ts";
 
 export type ResourceClient = PublicResourceClient;
 
+export interface ResourceClientChange {
+  readonly kind: "created" | "status-updated" | "deleted";
+  readonly resource: Readonly<Resource<unknown, unknown>>;
+}
+
 function deepFreeze<T>(value: T): T {
   if (!value || typeof value !== "object" || Object.isFrozen(value)) return value;
   Object.freeze(value);
@@ -18,16 +23,17 @@ function deepFreeze<T>(value: T): T {
 /** Restricts Resource reads and status writes to one PluginInstance namespace. */
 export function createResourceClient(
   store: PluginResourceStore,
-  onChange?: (resource: Readonly<Resource<unknown, unknown>>) => void,
+  onChange?: (change: ResourceClientChange) => void,
   assertCanCreateType?: (type: ResourceType) => void,
 ): ResourceClient {
   const changed = (
+    kind: ResourceClientChange["kind"],
     resource: Readonly<Resource<unknown, unknown>>,
   ): void => {
     try {
-      onChange?.(resource);
+      onChange?.(Object.freeze({ kind, resource }));
     } catch {
-      // Resource mutation has committed; projection invalidation must not turn it into failure.
+      // Resource mutation has committed; host reactions must not turn it into failure.
     }
   };
   const assertOwned = (
@@ -70,14 +76,14 @@ export function createResourceClient(
           spec: init.spec,
         }),
       );
-      changed(created);
+      changed("created", created);
       return created;
     },
     delete(type: ResourceType, name: string) {
       const resource = store.get(type, name);
       assertOwned(resource);
       store.delete(type, name);
-      changed(resource);
+      changed("deleted", resource);
     },
     patchStatus<TSpec, TStatus>(
       resource: Readonly<Resource<TSpec, TStatus>>,
@@ -98,7 +104,7 @@ export function createResourceClient(
         patched.metadata.resourceVersion !==
         resource.metadata.resourceVersion
       ) {
-        changed(patched);
+        changed("status-updated", patched);
       }
       return patched;
     },

tests/plugin-package.test.ts

@@ -184,6 +184,53 @@ describe("Plugin Package lifecycle", () => {
     await manager.close();
   });
 
+  test("automatically reconciles created Resources and status changes", async () => {
+    const root = testRoot();
+    const { instances, proposals } = stores(root);
+    instances.create({
+      pluginInstanceId: "reqloop_default",
+      pluginId: "qiankun/reqloop",
+      packageVersion: "1.2.0",
+    });
+    let context: PluginActivationContext | undefined;
+    const phases: string[] = [];
+    const manager = new Manager({
+      instances,
+      proposals,
+      packages: [
+        reqloopPackage((activation) => {
+          context = activation;
+          activation.registerController<
+            { requirement: string },
+            { phase?: string }
+          >({
+            resourceType: REQ_LOOP_RUN,
+            async reconcile(_baton, resource) {
+              phases.push(resource.status.phase ?? "pending");
+              if (resource.status.phase === undefined) {
+                activation.resources.patchStatus(resource, {
+                  phase: "ready",
+                });
+              }
+            },
+          });
+        }),
+      ],
+      onProposal() {},
+    });
+
+    await manager.start();
+    context!.resources.create(REQ_LOOP_RUN, {
+      name: "run_1",
+      spec: { requirement: "ship it" },
+    });
+
+    await waitFor(() => phases.length === 2);
+    expect(phases).toEqual(["pending", "ready"]);
+
+    await manager.close();
+  });
+
   test("presents active Resources as Board items and invalidates on status changes", async () => {
     const root = testRoot();
     const { instances, proposals } = stores(root);